From ff6c4fee1b00defea564b00d6fa9f1dc4fd92faa Mon Sep 17 00:00:00 2001 From: Bailey Dixon Date: Tue, 14 Jul 2026 14:28:45 -0400 Subject: [PATCH] fix: enforce agent wake boundaries --- CHANGELOG.md | 12 + DEVLOG.md | 36 +++ package.json | 2 +- src/server/audit.ts | 146 +++------ src/server/routers/__tests__/issue.test.ts | 57 ++++ src/server/routers/issue.ts | 2 + .../__tests__/agent-dispatch-inbox.test.ts | 48 +++ .../services/__tests__/agent-run.test.ts | 110 +++++++ .../__tests__/agent-wake-policy.test.ts | 75 +++++ src/server/services/__tests__/audit.test.ts | 276 +++++++++++++++++- src/server/services/__tests__/mcp.test.ts | 10 + .../services/__tests__/run-dispatcher.test.ts | 31 +- src/server/services/action-request-service.ts | 1 + src/server/services/agent-dispatch-inbox.ts | 22 +- src/server/services/agent-run.ts | 85 ++++-- src/server/services/agent-wake-policy.ts | 109 +++++++ src/server/services/ai-coach.ts | 2 + .../services/dispatch/run-dispatcher.ts | 111 ++++--- src/server/services/mcp.ts | 19 +- src/server/services/orchestration-service.ts | 4 + 20 files changed, 942 insertions(+), 216 deletions(-) create mode 100644 src/server/services/__tests__/agent-wake-policy.test.ts create mode 100644 src/server/services/agent-wake-policy.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f07ac58..4be7b459 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,18 @@ date, grouped by `Added` / `Changed` / `Fixed` / `Removed`. ## [Unreleased] +## [2026-07-14] — v0.13.1 · Reliable agent wake boundaries + +### Changed + +- **Activity and notifications no longer imply agent work.** Labels, ordinary status changes, non-escalating priority changes, stall and SLA markers, coach output, and other system activity remain visible without invoking an LLM. +- **Waiting agents resume through an explicit, engine-aware boundary.** Actionable human replies, direct mentions, explicit restarts, reassignment, and true priority escalation can wake work; RUNS agents receive a fresh provider turn while webhook and completions agents reactivate directly. + +### Fixed + +- **Agent output cannot wake its author.** Terminal and stalled replies, blocking markers, and automated comments no longer create replacement runs, while nudges produce one delivery instead of two. +- **Every terminal run leaves exactly one durable issue reply.** Completion, stall, and abandonment comments are claimed atomically so concurrent pollers cannot duplicate them. + ## [2026-07-14] — v0.13.0 · Reliable lifecycle operations ### Added diff --git a/DEVLOG.md b/DEVLOG.md index d24dc523..fabbd5a6 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -12728,3 +12728,39 @@ cover card content while scrolling. The repository-wide browser run passed 36 of 38 tests in parallel; the two failures (shared chat state and a composer locator exposed by the accessibility change) both passed independently after the composer name was corrected. + +--- + +## 2026-07-14 — Agent wake-boundary and terminal-comment audit + +Traced AXI-104 from production activity, delivery, and run records after a +stalled agent reply immediately opened a replacement run. The audit found that +agent-authored terminal comments were being judged by mutable payload fields +instead of authoritative actor identity; human status edits could directly +resume the assigned agent; all priority changes woke watchers; nudges emitted +two independently deliverable events; blocking and coach output could appear +operator-authored; and WAITING runs could be revived by incidental touches. + +Centralized wake policy now separates notification-only activity from +actionable work. Human BODY replies, safe agent-to-agent mentions, assignment, +explicit restart, and upward escalation into HIGH or URGENT are wakeable; +self-authored, actor-less, status, label, downward priority, stall, SLA, coach, +and blocking events remain durable activity without invoking an agent. Human +status changes no longer impersonate agent progress. The nudge marker and +AGENT_RUN_BLOCKED remain observable but do not create duplicate or bounce-back +deliveries. + +WAITING resume is now explicit and engine-aware. RUNS agents stay parked until +the dispatcher validates the canonical trigger and starts a fresh provider +turn containing the operator input; webhook and completions agents reactivate +for the actionable delivery. Completion, stall, and abandonment now use one +atomic terminal claim that creates exactly one agent comment even under +concurrent pollers and records its durable comment ID on the run. + +Verification: lint passed with existing repository warnings; typecheck passed; +the full Vitest suite passed 1,236 tests with 1 intentional live-connector skip; +the post-audit engine compatibility suites passed 28 tests; a production Next +build passed; and all 38 Playwright journeys passed serially. The canonical +gate first encountered the known cross-suite stale-work redispatch race; that +9-test file passed immediately in isolation and the full suite then passed +cleanly. diff --git a/package.json b/package.json index 2fbe788b..da419985 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "forge", - "version": "0.13.0", + "version": "0.13.1", "private": true, "description": "Forge — a fast, minimalist, keyboard-driven project management platform with pluggable agents.", "license": "MIT", diff --git a/src/server/audit.ts b/src/server/audit.ts index d4722a82..d901d52d 100644 --- a/src/server/audit.ts +++ b/src/server/audit.ts @@ -16,6 +16,12 @@ import { ALERTABLE_EVENT_KINDS, scheduleEventNotificationFanout, } from "@/server/services/notifications"; +import { + isActionableAssignedComment, + isActionableExplicitMention, + isActionablePriorityEscalation, + mentionedAgentIdsFromWakePayload, +} from "@/server/services/agent-wake-policy"; /** * Synthetic slug + url used to route agent-bound dispatches through the @@ -43,59 +49,7 @@ export function agentDispatchUrlFor(agentId: string): string { return `${AGENT_DISPATCH_WEBHOOK_URL_PREFIX}${agentId}`; } -function mentionedAgentIdsFromPayload(payloadIn: unknown): string[] { - const payload = payloadIn as - | { - agentRequests?: Array<{ agentId?: unknown }>; - mentions?: - | { - agentIds?: unknown[]; - agents?: Array<{ agentId?: unknown }>; - } - | Array<{ agentId?: unknown }>; - } - | undefined; - const out = new Set(); - if (Array.isArray(payload?.agentRequests)) { - for (const request of payload.agentRequests) { - if (typeof request.agentId === "string" && request.agentId.length > 0) { - out.add(request.agentId); - } - } - } - const mentionsRaw = payload?.mentions; - if (Array.isArray(mentionsRaw)) { - for (const mention of mentionsRaw) { - if (typeof mention.agentId === "string" && mention.agentId.length > 0) { - out.add(mention.agentId); - } - } - return [...out]; - } - if (mentionsRaw && typeof mentionsRaw === "object") { - if (Array.isArray(mentionsRaw.agentIds)) { - for (const id of mentionsRaw.agentIds) { - if (typeof id === "string" && id.length > 0) out.add(id); - } - } - if (Array.isArray(mentionsRaw.agents)) { - for (const mention of mentionsRaw.agents) { - if (typeof mention.agentId === "string" && mention.agentId.length > 0) { - out.add(mention.agentId); - } - } - } - } - return [...out]; -} - -const AGENT_WATCHER_FANOUT_EVENT_KINDS = new Set([ - EventKind.COMMENT_CREATED, - EventKind.ISSUE_PRIORITY_CHANGED, - EventKind.ISSUE_STALLED, - EventKind.ISSUE_SLA_BREACH, - EventKind.ISSUE_NUDGED, -]); +const AGENT_WATCHER_FANOUT_EVENT_KINDS = new Set([EventKind.COMMENT_CREATED]); const ENGAGEMENT_MODE_LABEL: Record = { EXECUTE: "EXECUTE", @@ -455,11 +409,7 @@ async function upsertAgentDispatchWebhook( EventKind.ISSUE_QUEUED, EventKind.COMMENT_CREATED, EventKind.ISSUE_PRIORITY_CHANGED, - EventKind.ISSUE_STALLED, - EventKind.ISSUE_SLA_BREACH, - EventKind.ISSUE_NUDGED, EventKind.CHAT_MESSAGE_POSTED, - EventKind.AGENT_RUN_BLOCKED, ], active: true, }, @@ -911,8 +861,7 @@ export async function recordChange( // to the assignee via a per-agent shim so the delivery carries its // own target id (doesn't re-resolve the issue's assignee later). if (params.eventKind === EventKind.ISSUE_PRIORITY_CHANGED && params.subjectType === "issue") { - const to = (params.payload as { to?: string } | undefined)?.to; - if (to === "HIGH" || to === "URGENT") { + if (isActionablePriorityEscalation(params.payload)) { const issue = await tx.issue.findUnique({ where: { id: params.subjectId }, select: { @@ -976,7 +925,7 @@ export async function recordChange( (params.subjectType === "issue" || params.subjectType === "execution-step"); if (isCommentMentionEvent) { // Normalize to a flat list of agent ids regardless of incoming shape. - const mentionedAgentIds = mentionedAgentIdsFromPayload(params.payload); + const mentionedAgentIds = [...mentionedAgentIdsFromWakePayload(params.payload)]; if (mentionedAgentIds.length) { // Load mentioned agents within workspace scope. We canonicalize // every mentioned active agent into the inbox so the operator's @@ -1006,6 +955,16 @@ export async function recordChange( }, }); for (const a of agents) { + if ( + !isActionableExplicitMention({ + actorId: params.actorId, + actorAgentId: params.actorAgentId ?? null, + targetAgentId: a.id, + payload: params.payload, + }) + ) { + continue; + } resolvedAgentIds.push(a.id); if (a.webhookUrl && !isRunsDrivenAgent(a)) { const wid = await upsertAgentDispatchWebhook( @@ -1053,47 +1012,10 @@ export async function recordChange( } } - // (d2) AGENT_RUN_BLOCKED — the agent itself just flipped a run to - // WAITING via `runs.setWaiting({ blocking: true })`. Route the - // event back to the agent's own webhook so the runtime sees its - // block confirmed (useful for receipts / acks). Subject is the - // agent-run row; we resolve the owning agent from the payload's - // `agentId`. Also surfaces the event in the standard agent - // activity feed via the normal ActivityEvent write that already - // happened above. - if (params.eventKind === EventKind.AGENT_RUN_BLOCKED && params.subjectType === "agent-run") { - const payload = params.payload as { agentId?: string } | undefined; - const blockedAgentId = - typeof payload?.agentId === "string" && payload.agentId.length > 0 ? payload.agentId : null; - if (blockedAgentId) { - const agent = await tx.agent.findFirst({ - where: { - workspaceId: params.workspaceId, - id: blockedAgentId, - archivedAt: null, - }, - select: { id: true, webhookUrl: true }, - }); - if (agent) { - resolvedAgentIds.push(agent.id); - if (agent.webhookUrl) { - const wid = await upsertAgentDispatchWebhook( - tx, - params.workspaceId, - agentDispatchUrlFor(agent.id), - ); - agentWebhookIds.push(wid); - } - } - } - } - - // (e) Watchers — for actionable issue-subject events, fan out to every - // subscribed agent watcher whose webhook is configured. Skip - // the actor's own row to avoid self-paging. Human watchers - // don't need a synthetic webhook — they're served by the - // inbox / notification surfaces. Agent watchers are routed - // through the same per-agent shim used for comment @mentions. + // (e) Watchers — human BODY comments can wake the currently assigned + // agent; explicit @mentions can wake another watched agent. Agent + // output requires an explicit mention and can never wake its author. + // Human watchers remain served by inbox / notification surfaces. // // Deliberately exclude low-signal lifecycle/status churn from the // durable agent inbox. A watched issue transitioning status or a @@ -1124,7 +1046,7 @@ export async function recordChange( !isCommentEdit && !isRollingStatusComment; if (isAgentWatcherFanoutEvent) { - const mentionedAgentIds = new Set(mentionedAgentIdsFromPayload(params.payload)); + const mentionedAgentIds = mentionedAgentIdsFromWakePayload(params.payload); const currentIssue = params.eventKind === EventKind.COMMENT_CREATED ? await tx.issue.findUnique({ @@ -1163,6 +1085,17 @@ export async function recordChange( }); for (const w of watchers) { if (!w.agent) continue; + if ( + params.eventKind === EventKind.COMMENT_CREATED && + !isActionableAssignedComment({ + actorId: params.actorId, + actorAgentId: params.actorAgentId ?? null, + targetAgentId: w.agent.id, + payload: params.payload, + }) + ) { + continue; + } // Body comments should wake the assigned agent and explicit @mentions. // Former assignees can remain sticky watchers; that stale watcher row // should not become fresh canonical work after reassignment. @@ -1176,15 +1109,6 @@ export async function recordChange( if (params.eventKind !== EventKind.COMMENT_CREATED && !w.wakeOnActivity) { continue; } - // Don't fan out the actor's own action back to themselves — - // when an agent comments and is also a watcher, the COMMENT_CREATED - // already routes via the assigned-agent + mention shims. Watcher - // fan-out is for OTHER stakeholders. - // Note: actorId is a User id, agentId is an Agent id; equality - // is impossible. We instead detect agent-as-actor via the - // payload's `agentId` field when present. - const payloadAgentId = (params.payload as { agentId?: string } | undefined)?.agentId; - if (payloadAgentId && payloadAgentId === w.agent.id) continue; resolvedAgentIds.push(w.agent.id); if (w.agent.webhookUrl && !isRunsDrivenAgent(w.agent)) { const wid = await upsertAgentDispatchWebhook( diff --git a/src/server/routers/__tests__/issue.test.ts b/src/server/routers/__tests__/issue.test.ts index dc3f09e7..054d6f46 100644 --- a/src/server/routers/__tests__/issue.test.ts +++ b/src/server/routers/__tests__/issue.test.ts @@ -102,6 +102,63 @@ describe("issueRouter — list filtering", () => { }); }); +describe("issueRouter — agent wake boundaries", () => { + it("records a human status transition without resuming the assigned agent", async () => { + const { caller, fixture } = await setup(); + const prisma = getPrisma(); + const issue = await createIssue(fixture, { title: "Metadata is not a prompt" }); + const currentIssue = await prisma.issue.findUniqueOrThrow({ + where: { id: issue.id }, + select: { statusId: true }, + }); + const agent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "Status worker", + profileKey: `status-worker-${Date.now()}`, + }, + }); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + const pausedAt = new Date(Date.now() - 60_000); + const run = await prisma.agentRun.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + status: AgentRunStatus.WAITING, + lastEventAt: pausedAt, + currentStep: "waiting for the operator", + }, + }); + const nextStatus = await prisma.status.findFirstOrThrow({ + where: { + workspaceId: fixture.workspace.id, + id: { not: currentIssue.statusId }, + category: { in: ["TODO", "IN_PROGRESS"] }, + }, + }); + + await caller.update({ id: issue.id, statusId: nextStatus.id }); + + const after = await prisma.agentRun.findUniqueOrThrow({ where: { id: run.id } }); + expect(after.status).toBe(AgentRunStatus.WAITING); + expect(after.lastEventAt).toEqual(pausedAt); + expect(after.currentStep).toBe("waiting for the operator"); + expect( + await prisma.activityEvent.count({ + where: { + workspaceId: fixture.workspace.id, + subjectId: issue.id, + kind: "ISSUE_STATUS_CHANGED", + }, + }), + ).toBe(1); + }); +}); + describe("issueRouter — archive integrity", () => { it("archives reversibly, closes live work, clears dispatch state, and audits both directions", async () => { const { caller, fixture } = await setup(); diff --git a/src/server/routers/issue.ts b/src/server/routers/issue.ts index 5739da8d..fd13d78b 100644 --- a/src/server/routers/issue.ts +++ b/src/server/routers/issue.ts @@ -1438,6 +1438,7 @@ export const issueRouter = router({ // Touch the run on transition by the assigned agent so the // status-change is reflected in the timeline as a STEP event. before.assignedAgentId && + ctx.apiKey?.linkedAgentId === before.assignedAgentId && patch.statusId && patch.statusId !== before.statusId ) { @@ -1449,6 +1450,7 @@ export const issueRouter = router({ payload: { from: before.statusId, to: patch.statusId, category: nextCategory }, actorId: ctx.session.user.id, actorAgentId: ctx.apiKey?.linkedAgentId ?? null, + resumeWaiting: true, }); } diff --git a/src/server/services/__tests__/agent-dispatch-inbox.test.ts b/src/server/services/__tests__/agent-dispatch-inbox.test.ts index 0ab819f9..1adea44e 100644 --- a/src/server/services/__tests__/agent-dispatch-inbox.test.ts +++ b/src/server/services/__tests__/agent-dispatch-inbox.test.ts @@ -100,6 +100,54 @@ describe("agent-dispatch-inbox — ensureCanonicalFromEvent", () => { expect(run!.assignmentEventId).toBe(run!.triggerEventId); }); + it("reactivates a webhook/completions WAITING run for an actionable human reply", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "INR" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await createAgent(fixture.workspace.id, "inr-a1"); + await prisma.agent.update({ + where: { id: agent.id }, + data: { runEngine: "COMPLETIONS" }, + }); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + const run = await prisma.agentRun.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + status: "WAITING", + }, + }); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Comment", + entityId: "inr-reply", + action: "create", + eventKind: EventKind.COMMENT_CREATED, + subjectType: "issue", + subjectId: issue.id, + payload: { commentId: "inr-reply", issueId: issue.id, kind: "BODY" }, + }); + + const after = await prisma.agentRun.findUniqueOrThrow({ where: { id: run.id } }); + expect(after.status).toBe("ACTIVE"); + expect(after.triggerKind).toBe(EventKind.COMMENT_CREATED); + }); + it("binds an ordinary materialized-issue assignment back to its single active step", async () => { const fixture = await createWorkspaceFixture({ keyPrefix: "INS" }); fixtures.push(fixture); diff --git a/src/server/services/__tests__/agent-run.test.ts b/src/server/services/__tests__/agent-run.test.ts index 2769a382..a9e7d713 100644 --- a/src/server/services/__tests__/agent-run.test.ts +++ b/src/server/services/__tests__/agent-run.test.ts @@ -166,6 +166,116 @@ describe("agent-run lifecycle", () => { ).toBe(1); }); + it("posts one stalled terminal comment without opening a replacement run", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "ARS" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await createAgent(fixture.workspace.id, "ars-a1"); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + const run = await prisma.agentRun.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + }, + }); + + await Promise.all([ + prisma.$transaction((tx) => + finishRun(tx, { + runId: run.id, + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + status: "STALLED", + summary: "Provider contract ended unexpectedly.", + }), + ), + prisma.$transaction((tx) => + finishRun(tx, { + runId: run.id, + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + status: "STALLED", + summary: "Provider contract ended unexpectedly.", + }), + ), + ]); + + const runs = await prisma.agentRun.findMany({ where: { issueId: issue.id } }); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe(AgentRunStatus.STALLED); + expect( + await prisma.comment.count({ + where: { + issueId: issue.id, + authoringAgentId: agent.id, + body: { contains: "[dispatch · run stalled]" }, + }, + }), + ).toBe(1); + expect( + await prisma.webhookDelivery.count({ + where: { event: { subjectType: "issue", subjectId: issue.id } }, + }), + ).toBe(0); + }); + + it("keeps an incidental touch parked until resume authority is explicit", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "ARP" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await createAgent(fixture.workspace.id, "arp-a1"); + const issue = await createIssue(fixture); + const pausedAt = new Date(Date.now() - 60_000); + const run = await prisma.agentRun.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + status: AgentRunStatus.WAITING, + lastEventAt: pausedAt, + }, + }); + + const parked = await prisma.$transaction((tx) => + openOrTouchRun(tx, { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + triggerEventId: "informational-touch", + triggerKind: EventKind.COMMENT_CREATED, + }), + ); + expect(parked.run.status).toBe(AgentRunStatus.WAITING); + expect(parked.run.lastEventAt).toEqual(pausedAt); + + const resumed = await prisma.$transaction((tx) => + openOrTouchRun(tx, { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + resumeWaiting: true, + }), + ); + expect(resumed.run.id).toBe(run.id); + expect(resumed.run.status).toBe(AgentRunStatus.ACTIVE); + expect(resumed.run.lastEventAt.getTime()).toBeGreaterThan(pausedAt.getTime()); + }); + it("a fresh assignment touch re-stamps mode + source and resumes WAITING (Phase 2)", async () => { // A genuine AGENT_ASSIGNED (fresh assignmentEventId) is an authoritative // (re)dispatch: the resolved mode the inbox passes in wins. Sticky-mode diff --git a/src/server/services/__tests__/agent-wake-policy.test.ts b/src/server/services/__tests__/agent-wake-policy.test.ts new file mode 100644 index 00000000..80c554c8 --- /dev/null +++ b/src/server/services/__tests__/agent-wake-policy.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import { + isActionableAssignedComment, + isActionableExplicitMention, + isActionablePriorityEscalation, +} from "@/server/services/agent-wake-policy"; + +describe("agent wake policy", () => { + it("only treats upward moves into high or urgent as priority wakes", () => { + expect(isActionablePriorityEscalation({ from: "MEDIUM", to: "HIGH" })).toBe(true); + expect(isActionablePriorityEscalation({ from: "HIGH", to: "URGENT" })).toBe(true); + expect(isActionablePriorityEscalation({ from: "HIGH", to: "LOW" })).toBe(false); + expect(isActionablePriorityEscalation({ from: "URGENT", to: "HIGH" })).toBe(false); + expect(isActionablePriorityEscalation({ from: "NONE", to: "MEDIUM" })).toBe(false); + }); + + it("separates human replies from self, agent, status, and system output", () => { + const targetAgentId = "agent-victor"; + expect( + isActionableAssignedComment({ + actorId: "user-bailey", + actorAgentId: null, + targetAgentId, + payload: { kind: "BODY" }, + }), + ).toBe(true); + expect( + isActionableAssignedComment({ + actorId: "key-owner", + actorAgentId: targetAgentId, + targetAgentId, + payload: { kind: "BODY" }, + }), + ).toBe(false); + expect( + isActionableAssignedComment({ + actorId: null, + actorAgentId: null, + targetAgentId, + payload: { kind: "BODY" }, + }), + ).toBe(false); + expect( + isActionableAssignedComment({ + actorId: "user-bailey", + actorAgentId: null, + targetAgentId, + payload: { kind: "STATUS" }, + }), + ).toBe(false); + }); + + it("requires an explicit mention for agent-to-agent wakes", () => { + const base = { + actorId: "key-owner", + actorAgentId: "agent-coach", + targetAgentId: "agent-victor", + }; + expect(isActionableAssignedComment({ ...base, payload: { kind: "BODY" } })).toBe(false); + expect( + isActionableAssignedComment({ + ...base, + payload: { kind: "BODY", mentions: { agentIds: ["agent-victor"] } }, + }), + ).toBe(true); + expect( + isActionableExplicitMention({ + actorId: "key-owner", + actorAgentId: "agent-victor", + targetAgentId: "agent-victor", + payload: { mentions: { agentIds: ["agent-victor"] } }, + }), + ).toBe(false); + }); +}); diff --git a/src/server/services/__tests__/audit.test.ts b/src/server/services/__tests__/audit.test.ts index c73e3f51..a3396bc4 100644 --- a/src/server/services/__tests__/audit.test.ts +++ b/src/server/services/__tests__/audit.test.ts @@ -10,9 +10,8 @@ import { } from "@/server/routers/__tests__/helpers"; /** - * Coverage for the audit.ts fan-out branches that emit SYSTEM comments - * (assignment notice) and route AGENT_RUN_BLOCKED to the owning agent's - * webhook. Real Postgres, no mocks. + * Coverage for assignment comments and the activity-event → agent-wake + * boundary. Real Postgres, no mocks. */ const fixtures: TestFixture[] = []; @@ -83,7 +82,9 @@ describe("audit.ts — AGENT_ASSIGNED system comment", () => { expect(sys[0].authoringAgentId).toBeNull(); expect(sys[0].body).toContain(`@${agent.profileKey}`); expect(sys[0].body).toContain(fixture.user.name ?? ""); - expect(sys[0].body).toContain("_Runtime: Test runtime · local daemon · tools from local daemon._"); + expect(sys[0].body).toContain( + "_Runtime: Test runtime · local daemon · tools from local daemon._", + ); // The SYSTEM comment is NOT routed through recordChange — verify // no COMMENT_CREATED row fired for it. @@ -357,6 +358,28 @@ describe("audit.ts — watcher fan-out", () => { }, }); expect(deliveries).toHaveLength(0); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Issue", + entityId: issue.id, + action: "set-labels", + eventKind: EventKind.ISSUE_UPDATED, + subjectType: "issue", + subjectId: issue.id, + payload: { labelIds: ["label-bug"], operation: "add" }, + }); + const labelEvent = await prisma.activityEvent.findFirstOrThrow({ + where: { subjectId: issue.id, kind: EventKind.ISSUE_UPDATED }, + orderBy: { createdAt: "desc" }, + }); + expect( + await prisma.agentRun.count({ + where: { issueId: issue.id, agentId: agent.id, triggerEventId: labelEvent.id }, + }), + ).toBe(0); + expect(await prisma.webhookDelivery.count({ where: { eventId: labelEvent.id } })).toBe(0); }); it("does not re-page watchers for rolling STATUS comment updates", async () => { @@ -499,6 +522,181 @@ describe("audit.ts — watcher fan-out", () => { expect(deliveries).toHaveLength(1); }); + it("does not let an agent-authored terminal comment wake its own replacement run", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "WFS" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "Victor", + profileKey: `wfs-victor-${Date.now()}`, + webhookUrl: "https://example.invalid/victor", + webhookSecret: "test-secret", + }, + }); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: null, + actorAgentId: agent.id, + entity: "Comment", + entityId: "terminal-comment-id", + action: "create", + eventKind: EventKind.COMMENT_CREATED, + subjectType: "issue", + subjectId: issue.id, + payload: { + commentId: "terminal-comment-id", + issueId: issue.id, + preview: "[dispatch · run stalled]", + // Intentionally omit payload.agentId: actorAgentId is authoritative. + }, + }); + + const event = await prisma.activityEvent.findFirstOrThrow({ + where: { + workspaceId: fixture.workspace.id, + subjectId: issue.id, + kind: EventKind.COMMENT_CREATED, + }, + }); + expect( + await prisma.agentRun.count({ + where: { workspaceId: fixture.workspace.id, issueId: issue.id }, + }), + ).toBe(0); + expect(await prisma.webhookDelivery.count({ where: { eventId: event.id } })).toBe(0); + }); + + it("does not treat actor-less automated comments as operator replies", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "WFA" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "Victor", + profileKey: `wfa-victor-${Date.now()}`, + webhookUrl: "https://example.invalid/victor", + webhookSecret: "test-secret", + }, + }); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: null, + entity: "Comment", + entityId: "automated-comment-id", + action: "create", + eventKind: EventKind.COMMENT_CREATED, + subjectType: "issue", + subjectId: issue.id, + payload: { commentId: "automated-comment-id", issueId: issue.id, kind: "BODY" }, + }); + + expect( + await prisma.agentRun.count({ + where: { workspaceId: fixture.workspace.id, issueId: issue.id }, + }), + ).toBe(0); + }); + + it("turns a nudge comment into one wake instead of dispatching its marker twice", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "WFN" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "Victor", + profileKey: `wfn-victor-${Date.now()}`, + webhookUrl: "https://example.invalid/victor", + webhookSecret: "test-secret", + }, + }); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Comment", + entityId: "nudge-comment-id", + action: "create", + eventKind: EventKind.COMMENT_CREATED, + subjectType: "issue", + subjectId: issue.id, + payload: { + commentId: "nudge-comment-id", + issueId: issue.id, + kind: "BODY", + isNudge: true, + mentions: { agentIds: [agent.id], userIds: [] }, + }, + }); + const commentEvent = await prisma.activityEvent.findFirstOrThrow({ + where: { subjectId: issue.id, kind: EventKind.COMMENT_CREATED }, + }); + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Issue", + entityId: issue.id, + action: "nudge", + eventKind: EventKind.ISSUE_NUDGED, + subjectType: "issue", + subjectId: issue.id, + payload: { commentId: "nudge-comment-id", agentId: agent.id }, + }); + const nudgeEvent = await prisma.activityEvent.findFirstOrThrow({ + where: { subjectId: issue.id, kind: EventKind.ISSUE_NUDGED }, + }); + + const runs = await prisma.agentRun.findMany({ where: { issueId: issue.id } }); + expect(runs).toHaveLength(1); + expect(runs[0]?.triggerEventId).toBe(commentEvent.id); + expect(await prisma.webhookDelivery.count({ where: { eventId: commentEvent.id } })).toBe(1); + expect(await prisma.webhookDelivery.count({ where: { eventId: nudgeEvent.id } })).toBe(0); + }); + it("does not open work for a former assigned-agent watcher when another agent is mentioned", async () => { const fixture = await createWorkspaceFixture({ keyPrefix: "WF4" }); fixtures.push(fixture); @@ -598,7 +796,7 @@ describe("audit.ts — watcher fan-out", () => { expect(deliveries[0]?.webhook.url).toBe(agentDispatchUrlFor(victor.id)); }); - it("only wakes agent watchers that are activity-wake enabled for generic issue activity", async () => { + it("does not wake agent watchers for a downward priority change", async () => { const fixture = await createWorkspaceFixture({ keyPrefix: "WF5" }); fixtures.push(fixture); const prisma = getPrisma(); @@ -639,6 +837,10 @@ describe("audit.ts — watcher fan-out", () => { ], }); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: active.id }, + }); await recordChange(prisma, { workspaceId: fixture.workspace.id, actorId: fixture.user.id, @@ -648,7 +850,7 @@ describe("audit.ts — watcher fan-out", () => { eventKind: EventKind.ISSUE_PRIORITY_CHANGED, subjectType: "issue", subjectId: issue.id, - payload: { priority: "HIGH" }, + payload: { from: "HIGH", to: "LOW" }, }); const event = await prisma.activityEvent.findFirstOrThrow({ @@ -666,7 +868,7 @@ describe("audit.ts — watcher fan-out", () => { triggerEventId: event.id, }, }); - expect(runs.map((run) => run.agentId)).toEqual([active.id]); + expect(runs).toHaveLength(0); const deliveries = await prisma.webhookDelivery.findMany({ where: { @@ -678,7 +880,65 @@ describe("audit.ts — watcher fan-out", () => { }, include: { webhook: true }, }); + expect(deliveries).toHaveLength(0); + }); + + it("wakes only the assigned agent once for a genuine priority escalation", async () => { + const fixture = await createWorkspaceFixture({ keyPrefix: "WF6" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const agent = await prisma.agent.create({ + data: { + workspaceId: fixture.workspace.id, + name: "Victor", + profileKey: `wf6-victor-${Date.now()}`, + webhookUrl: "https://example.invalid/victor", + webhookSecret: "test-secret", + }, + }); + const issue = await createIssue(fixture); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); + + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Issue", + entityId: issue.id, + action: "priority", + eventKind: EventKind.ISSUE_PRIORITY_CHANGED, + subjectType: "issue", + subjectId: issue.id, + payload: { from: "MEDIUM", to: "HIGH" }, + }); + + const event = await prisma.activityEvent.findFirstOrThrow({ + where: { + workspaceId: fixture.workspace.id, + kind: EventKind.ISSUE_PRIORITY_CHANGED, + subjectId: issue.id, + }, + }); + expect( + await prisma.agentRun.count({ + where: { issueId: issue.id, agentId: agent.id, triggerEventId: event.id }, + }), + ).toBe(1); + const deliveries = await prisma.webhookDelivery.findMany({ + where: { eventId: event.id }, + include: { webhook: true }, + }); expect(deliveries).toHaveLength(1); - expect(deliveries[0]?.webhook.url).toBe(agentDispatchUrlFor(active.id)); + expect(deliveries[0]?.webhook.url).toBe(agentDispatchUrlFor(agent.id)); }); }); diff --git a/src/server/services/__tests__/mcp.test.ts b/src/server/services/__tests__/mcp.test.ts index 92f15fbc..7bda0c50 100644 --- a/src/server/services/__tests__/mcp.test.ts +++ b/src/server/services/__tests__/mcp.test.ts @@ -4406,6 +4406,8 @@ describe("mcp runs.setWaiting + runs.resumeWork", () => { workspaceId: fixture.workspace.id, profileKey: `rw2-${Date.now()}`, name: "Patient2", + webhookUrl: "https://example.invalid/patient2", + webhookSecret: "test-secret", }, }); const issue = await createIssue(fixture); @@ -4438,6 +4440,14 @@ describe("mcp runs.setWaiting + runs.resumeWork", () => { const payload = blockedEvents[0].payload as Record; expect(payload.reason).toBe("review needed"); expect(payload.agentId).toBe(agent.id); + expect( + await prisma.webhookDelivery.count({ + where: { + eventId: blockedEvents[0].id, + webhook: { url: { startsWith: "agent:dispatch" } }, + }, + }), + ).toBe(0); }); it("setWaiting without linkedAgentId is rejected", async () => { diff --git a/src/server/services/__tests__/run-dispatcher.test.ts b/src/server/services/__tests__/run-dispatcher.test.ts index 23e37131..8d70d17b 100644 --- a/src/server/services/__tests__/run-dispatcher.test.ts +++ b/src/server/services/__tests__/run-dispatcher.test.ts @@ -428,6 +428,18 @@ describe("runs dispatcher", () => { select: { id: true }, }); const issue = await createIssue(fixture, { title: "blocked then replied" }); + await prisma.issue.update({ + where: { id: issue.id }, + data: { assignedAgentId: agent.id }, + }); + await prisma.issueWatcher.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + agentId: agent.id, + wakeOnActivity: true, + }, + }); // A run that paused 5 minutes ago via runs.setWaiting: WAITING, has an // externalRunId from the prior turn, nobody re-invoking it. const pausedAt = new Date(Date.now() - 5 * 60_000); @@ -443,7 +455,7 @@ describe("runs dispatcher", () => { }, }); // Operator reply that lands AFTER the run paused. - await prisma.comment.create({ + const reply = await prisma.comment.create({ data: { workspaceId: fixture.workspace.id, issueId: issue.id, @@ -451,6 +463,21 @@ describe("runs dispatcher", () => { body: "go ahead, skip the DB-backed tests", }, }); + await recordChange(prisma, { + workspaceId: fixture.workspace.id, + actorId: fixture.user.id, + entity: "Comment", + entityId: reply.id, + action: "create", + eventKind: EventKind.COMMENT_CREATED, + subjectType: "issue", + subjectId: issue.id, + payload: { commentId: reply.id, issueId: issue.id, kind: "BODY" }, + }); + + expect((await prisma.agentRun.findUniqueOrThrow({ where: { id: run.id } })).status).toBe( + AgentRunStatus.WAITING, + ); const tick = await ingestRunsDispatch(); expect(tick.resumed).toBeGreaterThanOrEqual(1); @@ -671,7 +698,7 @@ describe("runs dispatcher", () => { const after = await prisma.agentRun.findUniqueOrThrow({ where: { id: run.id } }); expect(after.status).toBe("STALLED"); expect(after.summary).toMatch(/without a valid Forge runs\.complete contract/); - expect(after.completionMeta).toBeNull(); + expect(after.completionMeta).toMatchObject({ terminalCommentId: expect.any(String) }); // A terminal failure must surface on the issue itself, not only in the // run overlay — posted as an agent-authored comment carrying the output. diff --git a/src/server/services/action-request-service.ts b/src/server/services/action-request-service.ts index f288d448..d5573eee 100644 --- a/src/server/services/action-request-service.ts +++ b/src/server/services/action-request-service.ts @@ -919,6 +919,7 @@ async function dispatchActionRequestKind( runEngine: grantEngine.engine, runEngineSource: grantEngine.source, runtimePolicy: runtimePolicy as unknown as Prisma.InputJsonValue, + resumeWaiting: true, }); // Collapse the runs we just abandoned (above) under the fresh grant // run so the stack reads as one thread. Prior STALLED attempts are diff --git a/src/server/services/agent-dispatch-inbox.ts b/src/server/services/agent-dispatch-inbox.ts index 89e735a6..dd82ef4c 100644 --- a/src/server/services/agent-dispatch-inbox.ts +++ b/src/server/services/agent-dispatch-inbox.ts @@ -186,9 +186,10 @@ const EMPTY_ENSURE: EnsureCanonicalResult = Object.freeze({ * Ensures canonical work rows exist for the just-recorded ActivityEvent. * Idempotent; callers can invoke once per `recordChange` write. * - * For issue-routed events (AGENT_ASSIGNED, ISSUE_QUEUED, mention - * COMMENT_CREATED, ISSUE_PRIORITY_CHANGED, watcher fan-out) we call - * `openOrTouchRun` once per resolved agent and stamp the trigger. + * For issue-routed events (AGENT_ASSIGNED, ISSUE_QUEUED, actionable + * COMMENT_CREATED, and true priority escalations) we call `openOrTouchRun` + * once per resolved agent and stamp the trigger. Notification-only issue + * activity never reaches this boundary. * * For chat-routed CHAT_MESSAGE_POSTED USER turns the user's ChatMessage * row IS the canonical work — we just ensure its lifecycle fields are @@ -362,13 +363,9 @@ async function ensureIssueRuns( // agent-request > payload > sticky-active-run > surface) mirrors the prior // `??`-chain exactly. // - // Behavioral equivalence with the deleted resolveIssueRunEngagementMode is - // exact, including its two EXECUTE fallbacks: (1) a non-assignee agent woken - // by a NON-comment event (watcher fan-out: ISSUE_STALLED / SLA / NUDGED / - // PRIORITY_CHANGED), and (2) a missing issue/workspace row. Both resolved to - // a bare EXECUTE before (after the upper tiers), regardless of agent binding - // or workspace default — so we force the surface default to EXECUTE for those - // cases here rather than letting the assignment surface read agent/ws config. + // Keep a defensive EXECUTE fallback for legacy non-assignee, non-comment + // canonical rows and for a missing issue/workspace. Current wake policy no + // longer creates those rows from stalled/SLA/nudge/metadata watcher churn. const forceExecuteDefault = !hasWorkspace || (!assigned && !commentWake); const surface: EngagementSurface = assigned ? "assignment" @@ -429,6 +426,11 @@ async function ensureIssueRuns( runEngineSource: engine?.source ?? null, runtimePolicy: runtimePolicy as Prisma.InputJsonValue | null, executionStepId, + // A fresh assignment is an explicit restart. Webhook/completions agents + // reactivate when their actionable wake is delivered. RUNS agents stay + // WAITING so resumeWaitingRuns can start a clean provider turn carrying + // the operator reply instead of polling the expired external run. + resumeWaiting: isAssigned || engine?.engine !== "RUNS", }); // Stamp the latest trigger on the run so the inbox can distinguish // "Victor was just re-mentioned in AXI-31" from "Victor is the diff --git a/src/server/services/agent-run.ts b/src/server/services/agent-run.ts index 6b7d59bf..40a368b5 100644 --- a/src/server/services/agent-run.ts +++ b/src/server/services/agent-run.ts @@ -74,9 +74,10 @@ function publishRunEvent(params: { * enforces it. * * "Non-terminal" includes both ACTIVE and WAITING. A WAITING run is a - * patient agent that's blocked on the operator; subsequent activity - * (operator nudge, MCP write) should resume it rather than open a - * fresh second run — `openOrTouchRun` handles the resume. + * patient agent that's blocked on the operator. Incidental output touches + * reuse it without resuming; only callers with explicit resume authority + * may flip it ACTIVE, so a held run never becomes work because metadata or + * an automated comment changed. */ export async function findActiveRun( tx: Tx, @@ -181,6 +182,14 @@ export async function openOrTouchRun( triggerEventId?: string | null; /** String mirror of ActivityEvent.kind for the latest wake. */ triggerKind?: string | null; + /** + * Explicit authority to move a parked WAITING run back to ACTIVE. + * Output/status touches leave this false; assignments, agent-owned + * transitions, and operator resume controls set it true. Human replies + * for RUNS engines stay WAITING until resumeWaitingRuns starts a fresh + * provider turn with the reply context. + */ + resumeWaiting?: boolean; }, ): Promise<{ run: AgentRun; isNew: boolean }> { const existing = await findActiveRun(tx, { @@ -209,7 +218,9 @@ export async function openOrTouchRun( // Conscious choice not to emit a separate event here — the // triggering caller (comment.create, MCP write, etc.) already // records its own audit row. - const resumeFromWaiting = existing.status === AgentRunStatus.WAITING; + const resumeAuthorized = params.resumeWaiting ?? Boolean(params.assignmentEventId); + const resumeFromWaiting = existing.status === AgentRunStatus.WAITING && resumeAuthorized; + const preserveParkedWaiting = existing.status === AgentRunStatus.WAITING && !resumeFromWaiting; // Re-arm run-budget enforcement when a paused run resumes. Inlined (not // imported from run-budget.ts) to avoid a module cycle — run-budget // already imports this file. No-op unless a budget breach/warn marker is @@ -238,7 +249,7 @@ export async function openOrTouchRun( const updated = await tx.agentRun.update({ where: { id: existing.id }, data: { - lastEventAt: new Date(), + ...(!preserveParkedWaiting ? { lastEventAt: new Date() } : {}), ...(resumeFromWaiting ? { status: AgentRunStatus.ACTIVE, controlState: "NONE", controlRequestedAt: null } : {}), @@ -503,32 +514,45 @@ export async function finishRun( }); } - let finished: AgentRun; - if (params.status === "COMPLETED") { - const completionMeta = - existing.completionMeta && - typeof existing.completionMeta === "object" && - !Array.isArray(existing.completionMeta) - ? { ...(existing.completionMeta as Record) } - : {}; - const knownCommentId = completionMeta.completionCommentId; - if (typeof knownCommentId !== "string" || !knownCommentId) { - const body = params.summary?.trim() || existing.summary?.trim() || "Run completed."; - const comment = await postAgentRunComment(tx, { - workspaceId: params.workspaceId, - issueId: params.issueId, - agentId: params.agentId, - body, - }); - completionMeta.completionCommentId = comment.id; - } - finished = await tx.agentRun.update({ - where: { id: params.runId }, - data: { completionMeta: completionMeta as Prisma.InputJsonValue }, + // Every terminal run gets exactly one durable issue comment. Keeping this + // inside the row-claim boundary makes failure comments idempotent too: two + // polling workers can observe the same provider terminal state, but only the + // winner above reaches this side effect. Previously the dispatcher posted a + // STALLED comment after finishRun returned, so a race could post twice and + // each comment could self-wake the same agent. + const completionMeta = + existing.completionMeta && + typeof existing.completionMeta === "object" && + !Array.isArray(existing.completionMeta) + ? { ...(existing.completionMeta as Record) } + : {}; + const knownTerminalCommentId = + typeof completionMeta.terminalCommentId === "string" + ? completionMeta.terminalCommentId + : params.status === "COMPLETED" && typeof completionMeta.completionCommentId === "string" + ? completionMeta.completionCommentId + : null; + if (!knownTerminalCommentId) { + const summary = params.summary?.trim() || existing.summary?.trim(); + const body = + params.status === "COMPLETED" + ? summary || "Run completed." + : params.status === "STALLED" + ? `[dispatch · run stalled]\n\n${summary || "The run stopped making progress before it completed."}` + : `[dispatch · run abandoned]\n\n${summary || "The run was stopped before it completed."}`; + const comment = await postAgentRunComment(tx, { + workspaceId: params.workspaceId, + issueId: params.issueId, + agentId: params.agentId, + body, }); - } else { - finished = await tx.agentRun.findUniqueOrThrow({ where: { id: params.runId } }); + completionMeta.terminalCommentId = comment.id; + if (params.status === "COMPLETED") completionMeta.completionCommentId = comment.id; } + const finished = await tx.agentRun.update({ + where: { id: params.runId }, + data: { completionMeta: completionMeta as Prisma.InputJsonValue }, + }); await tx.agentRunEvent.create({ data: { @@ -609,6 +633,7 @@ export async function postAgentRunComment( payload: { commentId: comment.id, issueId: params.issueId, + agentId: params.agentId, preview: params.body.slice(0, 120), }, }); @@ -637,6 +662,7 @@ export async function recordAgentAction( actorId?: string | null; actorAgentId?: string | null; assignmentEventId?: string | null; + resumeWaiting?: boolean; }, ): Promise<{ runId: string; isNewRun: boolean }> { const { run, isNew } = await openOrTouchRun(tx, { @@ -647,6 +673,7 @@ export async function recordAgentAction( actorAgentId: params.actorAgentId ?? null, assignmentEventId: params.assignmentEventId ?? null, currentStep: params.currentStep, + resumeWaiting: params.resumeWaiting ?? false, }); // Skip the "STARTED" duplicate event when the run was just opened — diff --git a/src/server/services/agent-wake-policy.ts b/src/server/services/agent-wake-policy.ts new file mode 100644 index 00000000..87588602 --- /dev/null +++ b/src/server/services/agent-wake-policy.ts @@ -0,0 +1,109 @@ +import "server-only"; + +/** + * Agent wake policy. + * + * Activity events are intentionally broader than agent work. Status, label, + * project, coach, and lifecycle events still belong in the audit timeline and + * notification system, but they must not become an implicit LLM prompt. Keep + * the small set of actionable wake decisions here so recordChange(), the RUNS + * dispatcher, and tests share one contract. + */ + +const PRIORITY_RANK: Readonly> = Object.freeze({ + NONE: 0, + LOW: 1, + MEDIUM: 2, + HIGH: 3, + URGENT: 4, +}); + +function asRecord(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} + +/** True only when priority moves upward into HIGH or URGENT. */ +export function isActionablePriorityEscalation(payload: unknown): boolean { + const record = asRecord(payload); + const from = typeof record?.from === "string" ? record.from : null; + const to = typeof record?.to === "string" ? record.to : null; + if (!from || !to || (to !== "HIGH" && to !== "URGENT")) return false; + return (PRIORITY_RANK[to] ?? -1) > (PRIORITY_RANK[from] ?? -1); +} + +/** Normalize structured and legacy mention payloads to agent ids. */ +export function mentionedAgentIdsFromWakePayload(payload: unknown): Set { + const out = new Set(); + const record = asRecord(payload); + + const requests = record?.agentRequests; + if (Array.isArray(requests)) { + for (const item of requests) { + const agentId = asRecord(item)?.agentId; + if (typeof agentId === "string" && agentId.length > 0) out.add(agentId); + } + } + + const mentions = record?.mentions; + if (Array.isArray(mentions)) { + for (const item of mentions) { + const agentId = asRecord(item)?.agentId; + if (typeof agentId === "string" && agentId.length > 0) out.add(agentId); + } + return out; + } + + const mentionRecord = asRecord(mentions); + const agentIds = mentionRecord?.agentIds; + if (Array.isArray(agentIds)) { + for (const agentId of agentIds) { + if (typeof agentId === "string" && agentId.length > 0) out.add(agentId); + } + } + const agents = mentionRecord?.agents; + if (Array.isArray(agents)) { + for (const item of agents) { + const agentId = asRecord(item)?.agentId; + if (typeof agentId === "string" && agentId.length > 0) out.add(agentId); + } + } + return out; +} + +/** + * Whether a BODY comment is new work for the assigned agent. + * + * - Human comments are actionable without requiring an @mention. + * - Agent-to-agent comments require an explicit @mention. + * - An agent can never wake itself. + * - Actor-less automation and rolling/system comments are informational. + */ +export function isActionableAssignedComment(params: { + actorId: string | null; + actorAgentId: string | null; + targetAgentId: string; + payload: unknown; +}): boolean { + const payload = asRecord(params.payload); + if (payload?.kind === "STATUS" || payload?.kind === "SYSTEM") return false; + if (params.actorAgentId === params.targetAgentId) return false; + + const mentioned = mentionedAgentIdsFromWakePayload(params.payload); + if (params.actorAgentId) return mentioned.has(params.targetAgentId); + if (params.actorId) return true; + return false; +} + +/** Explicit mentions use the same self-wake and actor provenance guard. */ +export function isActionableExplicitMention(params: { + actorId: string | null; + actorAgentId: string | null; + targetAgentId: string; + payload: unknown; +}): boolean { + if (!params.actorId && !params.actorAgentId) return false; + if (params.actorAgentId === params.targetAgentId) return false; + return mentionedAgentIdsFromWakePayload(params.payload).has(params.targetAgentId); +} diff --git a/src/server/services/ai-coach.ts b/src/server/services/ai-coach.ts index ed423df3..e442bf1a 100644 --- a/src/server/services/ai-coach.ts +++ b/src/server/services/ai-coach.ts @@ -154,6 +154,7 @@ export async function coachOnEvent( await recordChange(client, { workspaceId: params.workspaceId, actorId: null, + actorAgentId: coach.id, entity: "Comment", entityId: comment.id, action: "ai-coach", @@ -163,6 +164,7 @@ export async function coachOnEvent( subjectId: issue.id, payload: { commentId: comment.id, + agentId: coach.id, authoringAgentId: coach.id, coachEvent: params.eventKind, }, diff --git a/src/server/services/dispatch/run-dispatcher.ts b/src/server/services/dispatch/run-dispatcher.ts index e87db98a..a7a99fb9 100644 --- a/src/server/services/dispatch/run-dispatcher.ts +++ b/src/server/services/dispatch/run-dispatcher.ts @@ -3,12 +3,7 @@ import { AgentRunStatus, EventKind, Prisma } from "@prisma/client"; import type { AgentProvider } from "@prisma/client"; import { db } from "@/server/db"; import { logger } from "@/server/logger"; -import { - openOrTouchRun, - appendRunEvent, - finishRun, - postAgentRunComment, -} from "@/server/services/agent-run"; +import { openOrTouchRun, appendRunEvent, finishRun } from "@/server/services/agent-run"; import { FORGE_RUN_CONTRACT_VERSION, forgeRunInstruction } from "@/server/services/engagement-mode"; import { deriveRepoPath } from "@/server/services/repo-path"; import { buildRuntimePolicySnapshot, type RuntimePolicySnapshot } from "@/lib/runtime-enforcement"; @@ -29,6 +24,10 @@ import { resolveRunApproval, type PendingRunApproval, } from "@/server/services/run-approval-lifecycle"; +import { + isActionableAssignedComment, + isActionablePriorityEscalation, +} from "@/server/services/agent-wake-policy"; /** * Dispatch-via-runs ingestion (worker-hosted, poll-based). @@ -357,6 +356,7 @@ async function startNewRuns(): Promise { executionStepId: orchestrationContext?.step?.id ?? already?.executionStepId ?? null, engagementMode, orchestrationContext, + resumeWaiting: true, }); await tx.agentRun.update({ where: { id: run.id }, @@ -643,6 +643,8 @@ async function resumeWaitingRuns(limit: number): Promise { lastEventAt: true, runtimePolicy: true, completionMeta: true, + triggerEventId: true, + triggerKind: true, executionStepId: true, orchestrationContextSnapshot: true, issue: { @@ -685,29 +687,66 @@ async function resumeWaitingRuns(limit: number): Promise { }); if (!connector?.startRun) continue; - // A reply that landed after the run paused wakes it. Any comment the agent - // didn't author itself counts — the Nudge button and a plain `@agent` - // comment both produce one. Skip the agent's own posts so its closing - // status note doesn't self-resume. - const replies = await db.comment.findMany({ + // Resume only from the canonical trigger stamped by recordChange. Looking + // for "any newer comment" treated coach/system output as an operator + // reply and re-ran held work. The wake policy has already filtered the + // trigger once; validate it again at this provider boundary. + if (!run.triggerEventId || !run.triggerKind) continue; + const triggerEvent = await db.activityEvent.findFirst({ where: { - issueId: run.issueId, + id: run.triggerEventId, + workspaceId: run.workspaceId, createdAt: { gt: run.lastEventAt }, - deletedAt: null, - // Anything the waiting agent didn't author itself. Must include human - // comments (authoringAgentId IS NULL) — a bare `NOT: { authoringAgentId }` - // drops NULLs under SQL three-valued logic, so spell out the OR. - OR: [{ authoringAgentId: null }, { authoringAgentId: { not: run.agentId } }], - }, - orderBy: { createdAt: "asc" }, - take: 10, - select: { - body: true, - author: { select: { name: true } }, - authoringAgent: { select: { name: true } }, }, + select: { kind: true, actorId: true, actorAgentId: true, payload: true }, }); - if (replies.length === 0) continue; + if (!triggerEvent) continue; + + const triggerPayload = + triggerEvent.payload && + typeof triggerEvent.payload === "object" && + !Array.isArray(triggerEvent.payload) + ? (triggerEvent.payload as Record) + : {}; + let replyBlock: string | null = null; + if (triggerEvent.kind === EventKind.COMMENT_CREATED) { + if ( + !isActionableAssignedComment({ + actorId: triggerEvent.actorId, + actorAgentId: triggerEvent.actorAgentId, + targetAgentId: run.agentId, + payload: triggerPayload, + }) + ) { + continue; + } + const commentId = + typeof triggerPayload.commentId === "string" ? triggerPayload.commentId : null; + if (!commentId) continue; + const reply = await db.comment.findFirst({ + where: { + id: commentId, + workspaceId: run.workspaceId, + issueId: run.issueId, + kind: "BODY", + deletedAt: null, + createdAt: { gt: run.lastEventAt }, + }, + select: { + body: true, + author: { select: { name: true } }, + authoringAgent: { select: { name: true } }, + }, + }); + if (!reply) continue; + replyBlock = `${reply.authoringAgent?.name ?? reply.author?.name ?? "Operator"}: ${reply.body}`; + } else if ( + triggerEvent.kind === EventKind.ISSUE_PRIORITY_CHANGED && + isActionablePriorityEscalation(triggerPayload) + ) { + replyBlock = `Priority escalated from ${String(triggerPayload.from)} to ${String(triggerPayload.to)}.`; + } + if (!replyBlock) continue; const engagementMode = run.engagementMode; const instruction = forgeRunInstruction( @@ -730,9 +769,6 @@ async function resumeWaitingRuns(limit: number): Promise { }); const waitingReason = (run.currentStep || run.summary || "").trim(); - const replyBlock = replies - .map((r) => `${r.authoringAgent?.name ?? r.author?.name ?? "Operator"}: ${r.body}`) - .join("\n\n"); const orchestrationContext = await loadRunOrchestrationContext(db, { workspaceId: run.workspaceId, issueId: run.issueId, @@ -804,7 +840,8 @@ async function resumeWaitingRuns(limit: number): Promise { externalRunId, engine: "RUNS", resumedFromWaiting: true, - replyCount: replies.length, + triggerKind: triggerEvent.kind, + triggerEventId: run.triggerEventId, contractVersion: FORGE_RUN_CONTRACT_VERSION, }, }); @@ -1098,22 +1135,6 @@ async function pollActiveRuns(): Promise { status: terminal, summary, }); - // Surface a terminal *failure* on the issue itself. A clean - // COMPLETED run closed through `runs.complete`, so the agent has - // already posted its own comments via MCP — re-posting the summary - // would duplicate. But a STALLED/ABANDONED run's output otherwise - // lives only in `AgentRun.summary` (Mission Control overlay only), - // so without this a failed dispatch is invisible on the issue. Post - // it as an agent-authored comment so the timeline + watcher - // notifications carry it like any other reply. - if (terminal !== "COMPLETED" && run.issueId && summary) { - await postAgentRunComment(tx, { - workspaceId: run.workspaceId, - issueId: run.issueId, - agentId: run.agentId, - body: `[dispatch · run ${terminal.toLowerCase()}]\n\n` + summary, - }); - } const usage = status.usage; await tx.agentRun.update({ where: { id: run.id }, diff --git a/src/server/services/mcp.ts b/src/server/services/mcp.ts index 28044fee..70d91bf0 100644 --- a/src/server/services/mcp.ts +++ b/src/server/services/mcp.ts @@ -1896,6 +1896,7 @@ export const mcpTools = { actorId, actorAgentId: agentId, currentStep: `→ ${status.name}`, + resumeWaiting: true, }); if (!isNew) { await appendRunEvent(tx, { @@ -6473,6 +6474,7 @@ export const mcpTools = { actorAgentId: linkedAgentId, engagementMode: (input.mode as EngagementMode | undefined) ?? "EXECUTE", currentStep: input.summary ?? "opened by agent", + resumeWaiting: true, }), ); return { runId: run.id, isNew, status: run.status, issueId: issue.id }; @@ -6893,16 +6895,13 @@ export const mcpTools = { * resumes via `runs.resumeWork`, the stale timer measures * against the resume moment, not the original block moment. * - If `blocking: true`, also emits `AGENT_RUN_BLOCKED` via the - * standard `recordChange` flow. The audit fan-out branch in - * `audit.ts` then notifies the agent's webhook so its runtime - * sees the block confirmed. Set `blocking: false` (default) for - * silent waits that don't need to wake anyone. + * standard `recordChange` flow for operator alerts, activity, and + * notifications. It deliberately does not dispatch back to the agent + * that just parked itself. * - * Resuming work later: call `runs.resumeWork({ runId })`. Any - * subsequent activity routed through `openOrTouchRun` (a comment, - * a status upsert) will also auto-resume the run, so the explicit - * `resumeWork` call is only required for "I'm back" pings that - * don't otherwise touch the run. + * Resuming work later: call `runs.resumeWork({ runId })`, or let an + * actionable operator reply reach the dedicated RUNS resume dispatcher. + * Status output and unrelated issue changes leave the run parked. */ "runs.setWaiting": { scopes: ["WRITE_ISSUES"] as const, @@ -6919,7 +6918,7 @@ export const mcpTools = { .boolean() .default(false) .describe( - "When true, emit AGENT_RUN_BLOCKED so the agent's webhook is notified. Default false for silent waits.", + "When true, emit AGENT_RUN_BLOCKED for operator activity and notifications. It does not wake the parked agent.", ), }), async run(input: { runId: string; reason?: string; blocking: boolean }, ctx: McpContext) { diff --git a/src/server/services/orchestration-service.ts b/src/server/services/orchestration-service.ts index 1c21fb59..c9fe7013 100644 --- a/src/server/services/orchestration-service.ts +++ b/src/server/services/orchestration-service.ts @@ -1561,6 +1561,7 @@ export async function decomposeGoal( triggerEventId: event.id, triggerKind: EventKind.ISSUE_UPDATED, currentStep: `Planning: ${goal.title}`, + resumeWaiting: true, ...orchestrationRunStamp(plannerDispatchAgent, EngagementMode.DISCUSS), }); } else { @@ -2308,6 +2309,7 @@ async function transitionStepToReady( triggerKind: EventKind.EXECUTION_STEP_READY, currentStep: step.title, executionStepId: step.id, + resumeWaiting: true, ...orchestrationRunStamp(worker), }); } @@ -2531,6 +2533,7 @@ export async function retryExecutionStep( triggerKind: EventKind.EXECUTION_STEP_READY, currentStep: step.title, executionStepId: step.id, + resumeWaiting: true, ...orchestrationRunStamp(worker), }); @@ -2665,6 +2668,7 @@ export async function dispatchJudge( triggerKind: EventKind.EXECUTION_STEP_READY, currentStep: `Reviewing: ${step.title}`, executionStepId: step.id, + resumeWaiting: true, ...orchestrationRunStamp(judge, EngagementMode.REVIEW), }); const isRuntimeOnlyJudge = Boolean(judge.runtimeId && !judge.webhookUrl);