From aeffbe65eef5ca4a624c310a1cb7ce81af6838cd Mon Sep 17 00:00:00 2001 From: Chaim Marcus Date: Tue, 30 Jun 2026 10:18:38 +0000 Subject: [PATCH] =?UTF-8?q?feat(routing):=20at-least-once=20guarantee=20fo?= =?UTF-8?q?r=20chair=20comments=20=E2=86=92=20CEO?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The event-driven comment→CEO route (issue.comment.created) is best-effort: a failed/coalesced/cancelled CEO invoke, or a comment that lands while an agent is mid-run, can leave a chair comment unrouted with nothing to retry. This is exactly how a chair comment on CMP-190 (2026-06-30) was silently dropped — the host-native wake was cancelled (issue_execution_promoted / assignee_changed) and no plugin route fired. Adds a reconciliation sweep (route-pending-comments, every 2 min) that re-routes any active issue whose LAST comment is the chair's (no agent reply after it) and whose id ≠ the per-issue `last-routed-comment-id` marker. Both the fast path and the sweep stamp the marker → idempotent, no double-routing of healthy comments. "Last comment is user" precisely targets the failure (chair commented, no agent response) and skips historically answered comments, so first-run-after-deploy footprint is 0. - worker.ts: extract shared routeCommentToCeo() + markCommentRouted(); event handler stamps marker (incl. the native-handles-it branch); register route-pending-comments sweep with per-company/per-issue guards. - manifest.ts: declare route-pending-comments job (*/2 * * * *). Invariants: G12 (Paperclip touch only in the declared wrapper — plugin), G2 (extends the existing route, no parallel path). Refs legal-ai #164. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/manifest.ts | 7 ++ src/worker.ts | 211 +++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 188 insertions(+), 30 deletions(-) diff --git a/src/manifest.ts b/src/manifest.ts index c900536..9ca9e8f 100644 --- a/src/manifest.ts +++ b/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: [ { diff --git a/src/worker.ts b/src/worker.ts index 1311ba6..fc546ce 100644 --- a/src/worker.ts +++ b/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 { + 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 { + 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 - // rather than sending empty/error content as the instruction. - const promptBody = commentBody - ? commentBody - : "(לא ניתן לקרוא את תוכן התגובה — קרא את ה-comments האחרונים על ה-issue ישירות)"; - - try { - const { runId } = await ctx.agents.invoke(ceoAgentId, event.companyId, { - prompt: [ - `תגובה חדשה מחיים על issue "${issue.title}" (${issue.identifier || issueId}):`, - "", - 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), - }); - } + // Route through the shared helper (also used by the sweep). It stamps + // the `last-routed-comment-id` marker on success so the reconciliation + // sweep does not re-deliver this comment. + await routeCommentToCeo({ + issue, + companyId: event.companyId, + ceoAgentId, + commentBody: commentBody ?? "", + commentId, + source: "event", + }); }); // ── 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>; + 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>; + 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