Merge pull request 'feat(routing): ערובת at-least-once להערות-יו"ר → CEO (sweep-פיוס)' (#3) from feat/comment-delivery-guarantee into main

src/manifest.ts

@@ -178,6 +178,13 @@ export default {
 				'מסכם פידבק יו"ר מהשבוע האחרון ומעדכן את decision-lessons.md',
 			schedule: "0 19 * * 0",
 		},
+		{
+			jobKey: "route-pending-comments",
+			displayName: "ערובת מסירת-הערות (sweep)",
+			description:
+				"sweep פיוס: מנתב כל הערת-משתמש שטרם נותבה ל-CEO (at-least-once), כדי שהערה לא תיפול גם אם ה-wakeup התבטל/התלכד",
+			schedule: "*/2 * * * *",
+		},
 	],
 	webhooks: [
 		{

src/worker.ts

@@ -535,6 +535,83 @@ const plugin = definePlugin({
 
 		// Route user comments through CEO agent — per company
 
+		// Marker: the id of the last user comment already routed to the CEO for
+		// an issue. BOTH the event-driven route below and the reconciliation
+		// sweep (`route-pending-comments`) set it, so the sweep never re-routes a
+		// comment the fast path already handled. This is what turns best-effort
+		// event delivery into an at-least-once guarantee.
+		const LAST_ROUTED_COMMENT_KEY = "last-routed-comment-id";
+
+		async function markCommentRouted(
+			issueId: string,
+			commentId: string | undefined,
+		): Promise<void> {
+			if (!commentId) return;
+			try {
+				await ctx.state.set(
+					{
+						scopeKind: "issue",
+						scopeId: issueId,
+						stateKey: LAST_ROUTED_COMMENT_KEY,
+					},
+					commentId,
+				);
+			} catch (err) {
+				ctx.logger.warn("Failed to persist last-routed-comment marker", {
+					issueId,
+					commentId,
+					error: String(err),
+				});
+			}
+		}
+
+		// Invoke the company CEO to handle a user comment on an issue. Shared by
+		// the event-driven route and the reconciliation sweep. On success it
+		// stamps the marker so the comment is not routed twice. Returns true when
+		// the CEO was invoked.
+		async function routeCommentToCeo(args: {
+			issue: { id: string; title: string; identifier?: string | null };
+			companyId: string;
+			ceoAgentId: string;
+			commentBody: string;
+			commentId?: string;
+			source: "event" | "sweep";
+		}): Promise<boolean> {
+			const { issue, companyId, ceoAgentId, commentBody, commentId, source } =
+				args;
+			const promptBody = commentBody
+				? commentBody
+				: "(לא ניתן לקרוא את תוכן התגובה — קרא את ה-comments האחרונים על ה-issue ישירות)";
+			try {
+				const { runId } = await ctx.agents.invoke(ceoAgentId, companyId, {
+					prompt: [
+						`תגובה חדשה מחיים על issue "${issue.title}" (${issue.identifier || issue.id}):`,
+						"",
+						promptBody,
+						"",
+						`קרא את ה-comments האחרונים על ה-issue, הבן מה חיים מבקש, והחלט מה לעשות.`,
+						`אם ההוראה ברורה — נתב לסוכן המתאים. אם לא ברור — שאל את חיים.`,
+					].join("\n"),
+					reason: `user_commented_on_${issue.identifier || issue.id}`,
+				});
+				await markCommentRouted(issue.id, commentId);
+				ctx.logger.info("Routed user comment to CEO agent", {
+					issueId: issue.id,
+					commentId: commentId || null,
+					runId,
+					source,
+				});
+				return true;
+			} catch (err) {
+				ctx.logger.error("Failed to invoke CEO agent for comment routing", {
+					issueId: issue.id,
+					source,
+					error: String(err),
+				});
+				return false;
+			}
+		}
+
 		ctx.events.on("issue.comment.created", async (event) => {
 			// Only intercept human comments — not agent comments (prevents loops)
 			if (event.actorType !== "user") return;
@@ -601,6 +678,9 @@ const plugin = definePlugin({
 					"Comment reopened CEO-owned issue; native wake handles it, skipping plugin route",
 					{ issueId, ceoAgentId },
 				);
+				// The comment IS being handled (natively) — stamp the marker so the
+				// reconciliation sweep does not re-route it a second time.
+				await markCommentRouted(issueId, commentId);
 				return;
 			}
 
@@ -625,41 +705,25 @@ const plugin = definePlugin({
 						"(לא ניתן לקרוא את התגובה)";
 				} catch (err) {
 					ctx.logger.warn("listComments failed, falling back to bodySnippet", {
-						issueId, commentId, error: String(err),
+						issueId,
+						commentId,
+						error: String(err),
 					});
 					commentBody = payload?.bodySnippet ?? "";
 				}
 			}
 
-			// If we still have no body, instruct CEO to read comments directly
+			// Route through the shared helper (also used by the sweep). It stamps
-			// rather than sending empty/error content as the instruction.
+			// the `last-routed-comment-id` marker on success so the reconciliation
-			const promptBody = commentBody
+			// sweep does not re-deliver this comment.
-				? commentBody
+			await routeCommentToCeo({
-				: "(לא ניתן לקרוא את תוכן התגובה — קרא את ה-comments האחרונים על ה-issue ישירות)";
+				issue,
-
+				companyId: event.companyId,
-			try {
+				ceoAgentId,
-				const { runId } = await ctx.agents.invoke(ceoAgentId, event.companyId, {
+				commentBody: commentBody ?? "",
-					prompt: [
+				commentId,
-						`תגובה חדשה מחיים על issue "${issue.title}" (${issue.identifier || issueId}):`,
+				source: "event",
-						"",
-						promptBody,
-						"",
-						`קרא את ה-comments האחרונים על ה-issue, הבן מה חיים מבקש, והחלט מה לעשות.`,
-						`אם ההוראה ברורה — נתב לסוכן המתאים. אם לא ברור — שאל את חיים.`,
-					].join("\n"),
-					reason: `user_commented_on_${issue.identifier || issueId}`,
 			});
-				ctx.logger.info("Routed user comment to CEO agent", {
-					issueId,
-					commentId: commentId || null,
-					runId,
-				});
-			} catch (err) {
-				ctx.logger.error("Failed to invoke CEO agent for comment routing", {
-					issueId,
-					error: String(err),
-				});
-			}
 		});
 
 		// ── Jobs ───────────────────────────────────────────────────────
@@ -885,6 +949,93 @@ const plugin = definePlugin({
 			}
 		});
 
+		// Reconciliation sweep — the at-least-once guarantee for user comments.
+		// The event-driven route above is best-effort: a failed/coalesced CEO
+		// invoke, or a comment that lands while an agent is mid-run, can leave a
+		// user comment unrouted with nothing to retry (this is exactly how a
+		// chair comment was silently dropped — see legal-ai task #164). This sweep
+		// re-routes any issue whose newest user comment id does not match the
+		// `last-routed-comment-id` marker, so no comment falls through. Idempotent:
+		// the fast path stamps the marker, so a healthy comment is never
+		// double-routed. Runs every 2 minutes (manifest schedule).
+		ctx.jobs.register("route-pending-comments", async (_job) => {
+			let companies: Awaited<ReturnType<typeof ctx.companies.list>>;
+			try {
+				companies = await ctx.companies.list();
+			} catch (err) {
+				ctx.logger.warn("route-pending-comments: companies.list failed", {
+					error: String(err),
+				});
+				return;
+			}
+
+			for (const company of companies) {
+				const ceoAgentId = CEO_AGENT_IDS[company.id];
+				if (!ceoAgentId) continue;
+
+				let issues: Awaited<ReturnType<typeof ctx.issues.list>>;
+				try {
+					issues = await ctx.issues.list({ companyId: company.id, limit: 200 });
+				} catch (err) {
+					ctx.logger.warn("route-pending-comments: issues.list failed", {
+						companyId: company.id,
+						error: String(err),
+					});
+					continue;
+				}
+
+				for (const issue of issues) {
+					// Terminal issues need no routing. A fresh user comment on a
+					// `done` issue reopens it natively → it becomes active and is
+					// caught on the next sweep.
+					if (issue.status === "done" || issue.status === "cancelled") continue;
+
+					try {
+						const comments = await ctx.issues.listComments(
+							issue.id,
+							company.id,
+						);
+						if (comments.length === 0) continue;
+
+						// "Pending" = the conversation's last word is the chair's, with
+						// no agent reply after it. listComments is chronological (newest
+						// last). If an agent already responded, the comment was handled —
+						// skip. This both targets the exact failure (chair commented, no
+						// agent response) and avoids re-routing historically-answered
+						// comments on first run after deploy.
+						const last = comments[comments.length - 1];
+						if (last.authorType !== "user") continue;
+
+						const marker = await ctx.state.get({
+							scopeKind: "issue",
+							scopeId: issue.id,
+							stateKey: LAST_ROUTED_COMMENT_KEY,
+						});
+						// Already routed (CEO may still be working) — don't re-fire.
+						if (marker === last.id) continue;
+
+						ctx.logger.info(
+							"route-pending-comments: re-routing unrouted user comment",
+							{ issueId: issue.id, commentId: last.id },
+						);
+						await routeCommentToCeo({
+							issue,
+							companyId: company.id,
+							ceoAgentId,
+							commentBody: last.body,
+							commentId: last.id,
+							source: "sweep",
+						});
+					} catch (err) {
+						ctx.logger.warn("route-pending-comments: failed for issue", {
+							issueId: issue.id,
+							error: String(err),
+						});
+					}
+				}
+			}
+		});
+
 		// ── Data handlers (UI bridge) ──────────────────────────────────
 		// These back `usePluginData(key, params)` calls from the React UI bundle
 		// in `dist/ui/index.js`. Errors are caught and rethrown with a clear