feat(routing): at-least-once guarantee for chair comments → CEO

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) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 10:18:38 +00:00
parent 0a8790b088
commit aeffbe65ee
2 changed files with 188 additions and 30 deletions

View File

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

View File

@@ -535,6 +535,83 @@ const plugin = definePlugin({
// Route user comments through CEO agent — per company // 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) => { ctx.events.on("issue.comment.created", async (event) => {
// Only intercept human comments — not agent comments (prevents loops) // Only intercept human comments — not agent comments (prevents loops)
if (event.actorType !== "user") return; if (event.actorType !== "user") return;
@@ -601,6 +678,9 @@ const plugin = definePlugin({
"Comment reopened CEO-owned issue; native wake handles it, skipping plugin route", "Comment reopened CEO-owned issue; native wake handles it, skipping plugin route",
{ issueId, ceoAgentId }, { 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; return;
} }
@@ -625,41 +705,25 @@ const plugin = definePlugin({
"(לא ניתן לקרוא את התגובה)"; "(לא ניתן לקרוא את התגובה)";
} catch (err) { } catch (err) {
ctx.logger.warn("listComments failed, falling back to bodySnippet", { ctx.logger.warn("listComments failed, falling back to bodySnippet", {
issueId, commentId, error: String(err), issueId,
commentId,
error: String(err),
}); });
commentBody = payload?.bodySnippet ?? ""; 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 ─────────────────────────────────────────────────────── // ── 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) ────────────────────────────────── // ── Data handlers (UI bridge) ──────────────────────────────────
// These back `usePluginData(key, params)` calls from the React UI bundle // These back `usePluginData(key, params)` calls from the React UI bundle
// in `dist/ui/index.js`. Errors are caught and rethrown with a clear // in `dist/ui/index.js`. Errors are caught and rethrown with a clear