diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 6774731a73e..453e1205313 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -22,6 +22,8 @@ import { PreviewSnapshotToolkit, PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { MonitorToolkitHandlersLive } from "./toolkits/monitor/handlers.ts"; +import { MonitorToolkit } from "./toolkits/monitor/tools.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -208,10 +210,17 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( PreviewSnapshotRegistrationLive, ); +export const MonitorToolkitRegistrationLive = McpServer.toolkit(MonitorToolkit).pipe( + Layer.provide(MonitorToolkitHandlersLive), +); + const McpTransportLive = McpServer.layerHttp({ name: "T3 Code", version: packageJson.version, path: "/mcp", }).pipe(Layer.provide(McpAuthMiddlewareLive)); -export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive)); +export const layer = Layer.mergeAll( + PreviewToolkitRegistrationLive, + MonitorToolkitRegistrationLive, +).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index b13bf2d312e..98becb268ac 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -7,7 +7,7 @@ import { import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; -export type McpCapability = "preview"; +export type McpCapability = "preview" | "monitor"; export interface McpInvocationScope { readonly environmentId: EnvironmentId; @@ -25,7 +25,7 @@ export class McpInvocationContext extends Context.Service< >()("t3/mcp/McpInvocationContext") {} export const requireMcpCapability = Effect.fn("mcp.requireCapability")(function* ( - capability: McpCapability, + capability: "preview", ) { const invocation = yield* McpInvocationContext; if (!invocation.capabilities.has(capability)) { diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index 67c4f2f0ff0..d1b2a7e4ce4 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -114,7 +114,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), + capabilities: new Set(["preview", "monitor"]), issuedAt, expiresAt, }; diff --git a/apps/server/src/mcp/toolkits/monitor/handlers.ts b/apps/server/src/mcp/toolkits/monitor/handlers.ts new file mode 100644 index 00000000000..6d48bd0f2d1 --- /dev/null +++ b/apps/server/src/mcp/toolkits/monitor/handlers.ts @@ -0,0 +1,120 @@ +import { CommandId } from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as MonitorRegistry from "../../../monitor/MonitorRegistry.ts"; +import { cursorFromSnapshot } from "../../../monitor/monitorDiff.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { getPullRequestMonitorSnapshot } from "../../../sourceControl/gitHubPullRequestMonitor.ts"; +import { MonitorStartError, MonitorToolkit } from "./tools.ts"; + +const monitorStart = Effect.fn("MonitorToolkit.monitorStart")(function* ({ + prNumber, +}: { + readonly prNumber: number; +}) { + const invocation = yield* McpInvocationContext.McpInvocationContext; + if (!invocation.capabilities.has("monitor")) { + return yield* new MonitorStartError({ message: "PR monitoring is not available." }); + } + const registry = yield* MonitorRegistry.MonitorRegistry; + const existing = yield* registry.get(invocation.threadId); + if (Option.isSome(existing)) { + if (existing.value.prNumber !== prNumber) { + return yield* new MonitorStartError({ + message: `This thread is already monitoring PR #${existing.value.prNumber}. Ask the user before switching to a different pull request.`, + }); + } + return { + prNumber, + status: "monitoring" as const, + warning: null, + message: `PR #${prNumber} is already being monitored.`, + }; + } + + const snapshots = yield* ProjectionSnapshotQuery; + const readModel = yield* snapshots.getCommandReadModel(); + const thread = readModel.threads.find((candidate) => candidate.id === invocation.threadId); + const project = thread + ? readModel.projects.find((candidate) => candidate.id === thread.projectId) + : undefined; + if (!thread || !project) { + return yield* new MonitorStartError({ message: "The invoking thread no longer exists." }); + } + const repoCwd = thread.worktreePath ?? project.workspaceRoot; + const snapshot = yield* getPullRequestMonitorSnapshot({ + cwd: repoCwd, + pullRequestNumber: prNumber, + }).pipe(Effect.mapError((error) => new MonitorStartError({ message: error.message }))); + if (snapshot.state !== "open") { + return yield* new MonitorStartError({ + message: `PR #${prNumber} is ${snapshot.state}; only open pull requests can be monitored.`, + }); + } + + const createdAt = DateTime.formatIso(yield* DateTime.now); + const crypto = yield* Crypto.Crypto; + const commandId = CommandId.make(yield* crypto.randomUUIDv4); + const warning = snapshot.draft + ? `PR #${prNumber} is a draft. Monitoring started, but review bots may not run until it is marked ready for review.` + : null; + const generation = yield* registry.nextGeneration; + const won = yield* registry.registerIfAbsent({ + threadId: invocation.threadId, + prNumber, + generation, + startedAt: createdAt, + cursor: cursorFromSnapshot(snapshot), + wakeCount: 0, + repoCwd, + }); + if (!won) { + // A concurrent monitor_start beat us between the existence check and the + // registration write. Surface it rather than clobbering the winner. + return yield* new MonitorStartError({ + message: "A monitor was just started for this thread; check its status before retrying.", + }); + } + yield* OrchestrationEngineService.pipe( + Effect.flatMap((engine) => + engine.dispatch({ + type: "thread.monitor.start", + commandId, + threadId: invocation.threadId, + prNumber, + blockersSummary: warning ?? "", + headSha: snapshot.headSha, + createdAt, + }), + ), + Effect.mapError((error) => new MonitorStartError({ message: String(error) })), + // Roll back only the registration this call installed — a raced winner's + // registration must survive our failure. + Effect.onError(() => registry.removeGeneration(invocation.threadId, generation)), + ); + return { + prNumber, + status: "monitoring" as const, + warning, + message: warning ?? `Started monitoring PR #${prNumber}.`, + }; +}); + +export const MonitorToolkitHandlersLive = MonitorToolkit.toLayer({ + monitor_start: (input) => + monitorStart(input).pipe( + Effect.catch((error) => + typeof error === "object" && + error !== null && + "_tag" in error && + error._tag === "MonitorStartError" + ? error + : new MonitorStartError({ message: String(error) }), + ), + ), +}); diff --git a/apps/server/src/mcp/toolkits/monitor/tools.ts b/apps/server/src/mcp/toolkits/monitor/tools.ts new file mode 100644 index 00000000000..ad6c1b01021 --- /dev/null +++ b/apps/server/src/mcp/toolkits/monitor/tools.ts @@ -0,0 +1,41 @@ +import * as Schema from "effect/Schema"; +import * as Crypto from "effect/Crypto"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as MonitorRegistry from "../../../monitor/MonitorRegistry.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import * as GitHubCli from "../../../sourceControl/GitHubCli.ts"; + +export class MonitorStartError extends Schema.TaggedErrorClass()( + "MonitorStartError", + { message: Schema.String }, +) {} + +export const MonitorStartTool = Tool.make("monitor_start", { + description: + "Start monitoring a pull request for this thread. Call this after creating or identifying the PR when the user has asked you to monitor, babysit, or watch it for review feedback and CI results. The server will watch the PR and wake this thread when there is something to address. Never create the PR as a draft — review bots do not run on drafts.", + parameters: Schema.Struct({ prNumber: Schema.Int.check(Schema.isGreaterThan(0)) }), + success: Schema.Struct({ + prNumber: Schema.Number, + status: Schema.Literal("monitoring"), + warning: Schema.NullOr(Schema.String), + message: Schema.String, + }), + failure: MonitorStartError, + dependencies: [ + McpInvocationContext.McpInvocationContext, + MonitorRegistry.MonitorRegistry, + OrchestrationEngineService, + ProjectionSnapshotQuery, + GitHubCli.GitHubCli, + Crypto.Crypto, + ], +}) + .annotate(Tool.Title, "Start PR monitoring") + .annotate(Tool.Idempotent, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.OpenWorld, true); + +export const MonitorToolkit = Toolkit.make(MonitorStartTool); diff --git a/apps/server/src/monitor/MonitorRegistry.ts b/apps/server/src/monitor/MonitorRegistry.ts new file mode 100644 index 00000000000..cb52c095cfd --- /dev/null +++ b/apps/server/src/monitor/MonitorRegistry.ts @@ -0,0 +1,175 @@ +import { CommandId, type ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; + +import type { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import type { PullRequestMonitorCursor } from "./monitorDiff.ts"; + +export interface MonitorRegistration { + readonly threadId: ThreadId; + readonly prNumber: number; + readonly generation: number; + readonly startedAt: string; + readonly cursor: PullRequestMonitorCursor; + readonly wakeCount: number; + readonly repoCwd: string; +} + +export interface MonitorRegistryShape { + /** Registers only when the thread has no active registration; returns + whether this call won. Losing concurrent monitor_start calls must not + clobber the winner's cursor/generation. */ + readonly registerIfAbsent: (registration: MonitorRegistration) => Effect.Effect; + /** Removes only the exact generation the caller installed, so a failed + start can roll back without erasing a racing winner's registration. */ + readonly removeGeneration: (threadId: ThreadId, generation: number) => Effect.Effect; + readonly get: (threadId: ThreadId) => Effect.Effect>; + readonly updateCursor: ( + threadId: ThreadId, + cursor: PullRequestMonitorCursor, + expectedGeneration?: number, + ) => Effect.Effect; + readonly incrementWake: ( + threadId: ThreadId, + expectedGeneration?: number, + ) => Effect.Effect; + readonly setWakeCount: ( + threadId: ThreadId, + wakeCount: number, + expectedGeneration?: number, + ) => Effect.Effect; + readonly remove: ( + threadId: ThreadId, + expectedGeneration?: number, + ) => Effect.Effect>; + readonly listActive: Effect.Effect>; + readonly nextGeneration: Effect.Effect; +} + +export class MonitorRegistry extends Context.Service()( + "t3/monitor/MonitorRegistry", +) {} + +// Module-global store, mirroring McpProviderSession: the registry is +// constructed both in the runtime core (for the poller) and inside the MCP +// routes layer (for monitor_start), and those two layer instances must see +// the same registrations. All operations are synchronous, so plain-Map +// mutation inside Effect.sync is atomic per call. +const registrations = new Map(); +let generation = 0; + +const make: MonitorRegistryShape = { + registerIfAbsent: (registration) => + Effect.sync(() => { + if (registrations.has(registration.threadId)) return false; + registrations.set(registration.threadId, registration); + return true; + }), + removeGeneration: (threadId, generation) => + Effect.sync(() => { + const current = registrations.get(threadId); + if (current !== undefined && current.generation === generation) { + registrations.delete(threadId); + } + }), + get: (threadId) => Effect.sync(() => Option.fromNullishOr(registrations.get(threadId))), + updateCursor: (threadId, cursor, expectedGeneration) => + Effect.sync(() => { + const current = registrations.get(threadId); + if ( + current !== undefined && + (expectedGeneration === undefined || current.generation === expectedGeneration) + ) { + registrations.set(threadId, { ...current, cursor }); + } + }), + incrementWake: (threadId, expectedGeneration) => + Effect.sync(() => { + const current = registrations.get(threadId); + if ( + current === undefined || + (expectedGeneration !== undefined && current.generation !== expectedGeneration) + ) { + return current?.wakeCount ?? 0; + } + const wakeCount = (current?.wakeCount ?? 0) + 1; + registrations.set(threadId, { ...current, wakeCount }); + return wakeCount; + }), + setWakeCount: (threadId, wakeCount, expectedGeneration) => + Effect.sync(() => { + const current = registrations.get(threadId); + if ( + current !== undefined && + (expectedGeneration === undefined || current.generation === expectedGeneration) + ) { + registrations.set(threadId, { ...current, wakeCount }); + } + }), + remove: (threadId, expectedGeneration) => + Effect.sync(() => { + const current = registrations.get(threadId); + if ( + current === undefined || + (expectedGeneration !== undefined && current.generation !== expectedGeneration) + ) { + return Option.none(); + } + registrations.delete(threadId); + return Option.fromNullishOr(current); + }), + listActive: Effect.sync(() => [...registrations.values()]), + nextGeneration: Effect.sync(() => ++generation), +}; + +// The engine used for teardown dispatch is bound by PullRequestMonitor when +// it starts (it is the only construction site that has the engine and runs in +// the runtime core). Until then teardown leaves registrations intact because +// it cannot durably project the corresponding monitor end. +let activeEngine: OrchestrationEngineService["Service"] | undefined; + +export const bindEngine = (engine: OrchestrationEngineService["Service"]): void => { + activeEngine = engine; +}; + +export const layer = Layer.effect( + MonitorRegistry, + Effect.acquireRelease(Effect.succeed(MonitorRegistry.of(make)), () => + Effect.gen(function* () { + for (const registration of [...registrations.values()]) { + yield* endActiveMonitorForSession(registration.threadId); + } + }), + ), +); + +export const endActiveMonitorForSession = (threadId: ThreadId): Effect.Effect => + Effect.gen(function* () { + const registration = yield* make.get(threadId); + if (Option.isNone(registration) || activeEngine === undefined) return; + const endedAt = DateTime.formatIso(yield* DateTime.now); + const commandId = CommandId.make(`monitor-session-ended:${threadId}:${endedAt}`); + yield* activeEngine.dispatch({ + type: "thread.monitor.end", + commandId, + threadId, + reason: "session-ended", + blockersSummary: "", + endedAt, + }); + yield* make.remove(threadId, registration.value.generation); + }).pipe( + Effect.catch((error) => Effect.logWarning("monitor teardown failed", { threadId, error })), + ); + +export const __testing = { + make, + reset: (): void => { + registrations.clear(); + generation = 0; + activeEngine = undefined; + }, +}; diff --git a/apps/server/src/monitor/PullRequestMonitor.test.ts b/apps/server/src/monitor/PullRequestMonitor.test.ts new file mode 100644 index 00000000000..359733b3845 --- /dev/null +++ b/apps/server/src/monitor/PullRequestMonitor.test.ts @@ -0,0 +1,597 @@ +import { + CommandId, + EventId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCommand, + type OrchestrationThread, +} from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import { OrchestrationCommandInvariantError } from "../orchestration/Errors.ts"; +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { PersistenceSqlError } from "../persistence/Errors.ts"; +import { + GitHubPullRequestMonitorDecodeError, + type PullRequestMonitorSnapshot, +} from "../sourceControl/gitHubPullRequestMonitor.ts"; +import * as MonitorRegistry from "./MonitorRegistry.ts"; +import { cursorFromSnapshot } from "./monitorDiff.ts"; +import { make, PullRequestSnapshotFetcher } from "./PullRequestMonitor.ts"; + +const threadId = ThreadId.make("monitor-thread"); +const now = "2026-07-23T00:00:00.000Z"; + +const snapshot = ( + overrides: Partial = {}, +): PullRequestMonitorSnapshot => ({ + state: "open", + draft: false, + headSha: "head-1", + baseRefName: "main", + mergeability: "mergeable", + behindBaseBy: null, + requiredChecksKnown: true, + reviews: [], + reviewThreads: [], + issueComments: [], + checkRuns: [ + { + id: "ci-1", + name: "CI", + status: "in-progress", + conclusion: null, + startedAt: now, + headSha: "head-1", + }, + ], + ...overrides, +}); + +const comment = (id: string) => ({ + id, + author: { login: "bugbot", type: "app" as const }, + latestCommentByViewer: false, + body: `Finding ${id}`, + path: "src/file.ts", + line: 12, + createdAt: now, + updatedAt: now, + resolved: false, +}); + +const thread = (busy = false): OrchestrationThread => ({ + id: threadId, + projectId: ProjectId.make("project"), + title: "Monitor", + modelSelection: { instanceId: ProviderInstanceId.make("claude"), model: "default" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "feature", + worktreePath: "/repo", + latestTurn: busy + ? { + turnId: TurnId.make("busy-turn"), + state: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + assistantMessageId: null, + } + : null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + monitor: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, +}); + +const harness = Effect.fn("PullRequestMonitor.testHarness")(function* (input: { + readonly initial: PullRequestMonitorSnapshot; + readonly current: { value: PullRequestMonitorSnapshot }; + readonly wakeCount?: number; + readonly failWake?: boolean; + readonly failEnd?: boolean; + readonly failFreshFetch?: boolean; + readonly reconcileFailures?: number; + readonly reconcileThreadId?: ThreadId; + readonly busy?: { value: boolean }; + readonly afterUpdate?: () => void; + readonly changeGenerationAfterUpdate?: boolean; +}) { + let registration: MonitorRegistry.MonitorRegistration | undefined = { + threadId, + prNumber: 42, + generation: 1, + startedAt: now, + cursor: cursorFromSnapshot(input.initial), + wakeCount: input.wakeCount ?? 0, + repoCwd: "/repo", + }; + const commands: OrchestrationCommand[] = []; + const ordering: string[] = []; + const registry = MonitorRegistry.MonitorRegistry.of({ + registerIfAbsent: (next) => + Effect.sync(() => { + if (registration !== undefined) return false; + registration = next; + return true; + }), + removeGeneration: (_, generation) => + Effect.sync(() => { + if (registration?.generation === generation) registration = undefined; + }), + get: (requestedThreadId) => + Effect.sync(() => + registration?.threadId === requestedThreadId ? Option.some(registration) : Option.none(), + ), + updateCursor: (_, cursor, expectedGeneration) => + Effect.sync(() => { + ordering.push("ack"); + if ( + registration && + (expectedGeneration === undefined || registration.generation === expectedGeneration) + ) { + registration = { ...registration, cursor }; + } + }), + incrementWake: (_, expectedGeneration) => + Effect.sync(() => { + if ( + registration && + expectedGeneration !== undefined && + registration.generation !== expectedGeneration + ) { + return registration.wakeCount; + } + const count = (registration?.wakeCount ?? 0) + 1; + if (registration) registration = { ...registration, wakeCount: count }; + return count; + }), + setWakeCount: (_, wakeCount, expectedGeneration) => + Effect.sync(() => { + if ( + registration && + (expectedGeneration === undefined || registration.generation === expectedGeneration) + ) { + registration = { ...registration, wakeCount }; + } + }), + remove: (_, expectedGeneration) => + Effect.sync(() => { + if ( + registration && + expectedGeneration !== undefined && + registration.generation !== expectedGeneration + ) { + return Option.none(); + } + const removed = registration ? Option.some(registration) : Option.none(); + registration = undefined; + return removed; + }), + listActive: Effect.sync(() => (registration ? [registration] : [])), + nextGeneration: Effect.succeed(2), + }); + const engine = OrchestrationEngineService.of({ + readEvents: () => Stream.empty, + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + dispatch: (command) => { + commands.push(command); + if (command.type === "thread.monitor.update") { + input.afterUpdate?.(); + if (input.changeGenerationAfterUpdate && registration) { + registration = { ...registration, generation: registration.generation + 1 }; + } + } + if (command.type === "thread.turn.start") { + ordering.push("dispatch"); + if (input.failWake) { + return Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "rejected", + }), + ); + } + } + if (command.type === "thread.monitor.end" && input.failEnd) { + return Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "rejected", + }), + ); + } + return Effect.succeed({ sequence: commands.length }); + }, + }); + let fetchCount = 0; + let reconcileCount = 0; + const monitor = yield* make.pipe( + Effect.provideService(MonitorRegistry.MonitorRegistry, registry), + Effect.provideService(PullRequestSnapshotFetcher, { + fetch: () => { + fetchCount += 1; + return input.failFreshFetch && fetchCount === 2 + ? Effect.fail( + new GitHubPullRequestMonitorDecodeError({ + command: "gh", + cwd: "/repo", + detail: "fresh fetch failed", + cause: null, + }), + ) + : Effect.succeed(input.current.value); + }, + }), + Effect.provideService(OrchestrationEngineService, engine), + Effect.provide( + Layer.mergeAll( + Layer.mock(ProjectionSnapshotQuery)({ + getShellSnapshot: () => { + reconcileCount += 1; + if (reconcileCount <= (input.reconcileFailures ?? 0)) { + return Effect.fail( + new PersistenceSqlError({ + operation: "PullRequestMonitor.test:getShellSnapshot", + detail: "reconcile failed", + }), + ); + } + return Effect.succeed({ + snapshotSequence: 0, + projects: [], + threads: input.reconcileThreadId + ? [ + { + ...thread(), + id: input.reconcileThreadId, + monitor: { + status: "monitoring" as const, + prNumber: 42, + startedAt: now, + blockersSummary: "", + headSha: "head-1", + wakeCount: 0, + updatedAt: now, + endedAt: null, + endedReason: null, + }, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + }, + ] + : [], + updatedAt: now, + }); + }, + getThreadDetailById: () => + Effect.succeed(Option.some(thread(input.busy?.value ?? false))), + }), + NodeServices.layer, + ), + ), + ); + return { monitor, commands, ordering, getRegistration: () => registration }; +}); + +const endReason = (commands: ReadonlyArray) => + commands.find((command) => command.type === "thread.monitor.end"); + +describe("PullRequestMonitor dispatch protocol", () => { + it.effect("terminal-discards-cycle", () => + Effect.gen(function* () { + const initial = snapshot(); + const h = yield* harness({ + initial, + current: { value: snapshot({ state: "merged", reviewThreads: [comment("late")] }) }, + }); + yield* h.monitor.pollOnce; + assert.strictEqual(endReason(h.commands)?.type, "thread.monitor.end"); + assert.strictEqual( + h.commands.some((command) => command.type === "thread.turn.start"), + false, + ); + }), + ); + + it.effect("ready-ends after a final update", () => + Effect.gen(function* () { + const ready = snapshot({ + checkRuns: [{ ...snapshot().checkRuns[0]!, status: "completed", conclusion: "success" }], + }); + const h = yield* harness({ initial: snapshot(), current: { value: ready } }); + yield* h.monitor.pollOnce; + assert.deepStrictEqual( + h.commands.map((command) => command.type), + ["thread.monitor.update", "thread.monitor.end"], + ); + }), + ); + + it.effect("ignores unresolved review threads that predate monitoring", () => + Effect.gen(function* () { + const oldComment = { + ...comment("old"), + createdAt: "2026-07-22T00:00:00.000Z", + }; + const ready = snapshot({ + reviewThreads: [oldComment], + checkRuns: [{ ...snapshot().checkRuns[0]!, status: "completed", conclusion: "success" }], + }); + const h = yield* harness({ initial: snapshot(), current: { value: ready } }); + yield* h.monitor.pollOnce; + assert.strictEqual(endReason(h.commands)?.type, "thread.monitor.end"); + }), + ); + + it.effect("wake-then-ack ordering", () => + Effect.gen(function* () { + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + }); + yield* h.monitor.pollOnce; + assert.deepStrictEqual(h.ordering, ["dispatch", "ack"]); + assert.strictEqual(h.getRegistration()?.wakeCount, 1); + }), + ); + + it.effect("failed-dispatch-preserves-cursor", () => + Effect.gen(function* () { + const initial = snapshot(); + const h = yield* harness({ + initial, + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + failWake: true, + }); + yield* h.monitor.pollOnce; + assert.deepStrictEqual(h.getRegistration()?.cursor, cursorFromSnapshot(initial)); + assert.strictEqual(h.getRegistration()?.wakeCount, 0); + }), + ); + + it.effect("failed-send-boundary-fetch-discards-wake-without-acking", () => + Effect.gen(function* () { + const initial = snapshot(); + const h = yield* harness({ + initial, + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + failFreshFetch: true, + }); + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.some((command) => command.type === "thread.turn.start"), + false, + ); + assert.deepStrictEqual(h.getRegistration()?.cursor, cursorFromSnapshot(initial)); + assert.strictEqual(h.getRegistration()?.wakeCount, 0); + }), + ); + + it.effect("provider failure correlates by dispatch time and rewinds cursor plus wake count", () => + Effect.gen(function* () { + const initial = snapshot(); + const h = yield* harness({ + initial, + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + }); + yield* h.monitor.pollOnce; + const wake = h.commands.find((command) => command.type === "thread.turn.start"); + assert.ok(wake?.type === "thread.turn.start"); + const failureEvent = (createdAt: string) => + ({ + sequence: 1, + eventId: EventId.make(`failure-${createdAt}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: createdAt, + commandId: CommandId.make(`failure-${createdAt}`), + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.activity-appended", + payload: { + threadId, + activity: { + id: EventId.make(`activity-${createdAt}`), + tone: "error", + kind: "provider.turn.start.failed", + summary: "Provider turn start failed", + payload: {}, + turnId: null, + createdAt, + }, + }, + }) as const; + yield* h.monitor.handleDomainEvent(failureEvent("0001-01-01T00:00:00.000Z")); + assert.strictEqual(h.getRegistration()?.wakeCount, 1); + yield* h.monitor.handleDomainEvent(failureEvent(wake.createdAt)); + assert.strictEqual(h.getRegistration()?.wakeCount, 0); + assert.deepStrictEqual(h.getRegistration()?.cursor, cursorFromSnapshot(initial)); + }), + ); + + it.effect("user-turn-discards-pending-wake", () => + Effect.gen(function* () { + const busy = { value: true }; + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + busy, + }); + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.some((command) => command.type === "thread.turn.start"), + false, + ); + assert.deepStrictEqual(h.getRegistration()?.cursor, cursorFromSnapshot(snapshot())); + }), + ); + + it.effect("generation-change-discards", () => + Effect.gen(function* () { + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + changeGenerationAfterUpdate: true, + }); + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.some((command) => command.type === "thread.turn.start"), + false, + ); + }), + ); + + it.effect("coalesces events while a wake is in flight", () => + Effect.gen(function* () { + const current = { value: snapshot({ reviewThreads: [comment("one")] }) }; + const h = yield* harness({ initial: snapshot(), current }); + yield* h.monitor.pollOnce; + current.value = snapshot({ + reviewThreads: [comment("one"), comment("two"), comment("three")], + }); + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.filter((command) => command.type === "thread.turn.start").length, + 1, + ); + yield* h.monitor.handleDomainEvent({ + sequence: 1, + eventId: EventId.make("session-ready"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("session-ready"), + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.session-set", + payload: { + threadId, + session: { + threadId, + status: "ready", + providerName: "claude", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + }); + yield* h.monitor.pollOnce; + const wakes = h.commands.filter((command) => command.type === "thread.turn.start"); + assert.strictEqual(wakes.length, 2); + assert.match(wakes[1]!.message.text, /Finding two/); + assert.match(wakes[1]!.message.text, /Finding three/); + }), + ); + + it.effect("breaker-at-10", () => + Effect.gen(function* () { + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot({ reviewThreads: [comment("one")] }) }, + wakeCount: 10, + }); + yield* h.monitor.pollOnce; + const ended = endReason(h.commands); + assert.strictEqual(ended?.type === "thread.monitor.end" && ended.reason, "needs-attention"); + }), + ); + + it.effect("archives by projecting stopped before removing the registration", () => + Effect.gen(function* () { + const h = yield* harness({ initial: snapshot(), current: { value: snapshot() } }); + yield* h.monitor.handleDomainEvent({ + sequence: 1, + eventId: EventId.make("archived"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("archived"), + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.archived", + payload: { threadId, archivedAt: now, updatedAt: now }, + }); + const ended = endReason(h.commands); + assert.strictEqual(ended?.type === "thread.monitor.end" && ended.reason, "stopped"); + assert.strictEqual(h.getRegistration(), undefined); + }), + ); + + it.effect("removes an archived registration when projecting monitor end fails", () => + Effect.gen(function* () { + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot() }, + failEnd: true, + }); + yield* h.monitor.handleDomainEvent({ + sequence: 1, + eventId: EventId.make("archived"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("archived"), + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.archived", + payload: { threadId, archivedAt: now, updatedAt: now }, + }); + assert.strictEqual(h.getRegistration(), undefined); + }), + ); + + it.effect("retries boot reconciliation on the first poll after a failed attempt", () => + Effect.gen(function* () { + const orphanThreadId = ThreadId.make("orphan-monitor-thread"); + const h = yield* harness({ + initial: snapshot(), + current: { value: snapshot() }, + reconcileFailures: 1, + reconcileThreadId: orphanThreadId, + }); + + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.some( + (command) => command.type === "thread.monitor.end" && command.threadId === orphanThreadId, + ), + false, + ); + + yield* h.monitor.pollOnce; + assert.strictEqual( + h.commands.some( + (command) => command.type === "thread.monitor.end" && command.threadId === orphanThreadId, + ), + true, + ); + }), + ); +}); diff --git a/apps/server/src/monitor/PullRequestMonitor.ts b/apps/server/src/monitor/PullRequestMonitor.ts new file mode 100644 index 00000000000..291673c2cba --- /dev/null +++ b/apps/server/src/monitor/PullRequestMonitor.ts @@ -0,0 +1,461 @@ +import { CommandId, MessageId, type OrchestrationEvent, type ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as GitHubCli from "../sourceControl/GitHubCli.ts"; + +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + getPullRequestMonitorSnapshot, + type GitHubPullRequestMonitorError, + type PullRequestMonitorSnapshot, +} from "../sourceControl/gitHubPullRequestMonitor.ts"; +import * as MonitorRegistry from "./MonitorRegistry.ts"; +import { diffPullRequestMonitorSnapshot, type PullRequestMonitorCursor } from "./monitorDiff.ts"; +import { computeReadiness } from "./readiness.ts"; +import { buildWakePrompt, formatBlockersSummary } from "./wakePrompt.ts"; + +const POLL_INTERVAL = Duration.seconds(30); +const MAX_BACKOFF = Duration.minutes(5); + +interface WakeClaim { + readonly generation: number; + readonly commandId: CommandId; + readonly phase: "pending" | "in-flight"; + readonly dispatchedAt?: string; + readonly previousWakeCount: number; + // Cursor as it was before this wake acked, so a wake that dies after the + // ack — provider start failure, superseded by a user turn — can rewind and + // let the next poll re-diff the same events instead of losing them (I5). + readonly previousCursor: PullRequestMonitorCursor; +} + +export interface PullRequestSnapshotFetcherShape { + readonly fetch: (input: { + readonly cwd: string; + readonly pullRequestNumber: number; + }) => Effect.Effect; +} + +export class PullRequestSnapshotFetcher extends Context.Service< + PullRequestSnapshotFetcher, + PullRequestSnapshotFetcherShape +>()("t3/monitor/PullRequestMonitor/PullRequestSnapshotFetcher") {} + +export const PullRequestSnapshotFetcherLive = Layer.effect( + PullRequestSnapshotFetcher, + Effect.gen(function* () { + const githubCli = yield* GitHubCli.GitHubCli; + return PullRequestSnapshotFetcher.of({ + fetch: (input) => + getPullRequestMonitorSnapshot(input).pipe( + Effect.provideService(GitHubCli.GitHubCli, githubCli), + ), + }); + }), +); + +export class PullRequestMonitor extends Context.Service< + PullRequestMonitor, + { + readonly start: () => Effect.Effect; + readonly pollOnce: Effect.Effect; + readonly handleDomainEvent: (event: OrchestrationEvent) => Effect.Effect; + } +>()("t3/monitor/PullRequestMonitor") {} + +const threadIdOf = (event: OrchestrationEvent): ThreadId | undefined => + "threadId" in event.payload ? event.payload.threadId : undefined; + +export const make = Effect.gen(function* () { + const registry = yield* MonitorRegistry.MonitorRegistry; + const fetcher = yield* PullRequestSnapshotFetcher; + const engine = yield* OrchestrationEngineService; + // Give session-teardown paths (which run outside this service's context) a + // way to dispatch thread.monitor.end through the real engine. + MonitorRegistry.bindEngine(engine); + const snapshots = yield* ProjectionSnapshotQuery; + const crypto = yield* Crypto.Crypto; + const claims = yield* Ref.make>(new Map()); + const lastPollFailed = yield* Ref.make(false); + const reconciled = yield* Ref.make(false); + + const mutateClaim = ( + threadId: ThreadId, + f: (claim: WakeClaim | undefined) => WakeClaim | undefined, + ) => + Ref.update(claims, (current) => { + const next = new Map(current); + const claim = f(next.get(threadId)); + if (claim === undefined) next.delete(threadId); + else next.set(threadId, claim); + return next; + }); + + // Drop a claim matching `shouldRelease`; if its wake had already advanced + // the cursor, rewind to the pre-wake cursor so those events re-diff (I5). + const releaseClaim = (threadId: ThreadId, shouldRelease: (claim: WakeClaim) => boolean) => + Ref.modify(claims, (current) => { + const claim = current.get(threadId); + if (claim === undefined || !shouldRelease(claim)) { + return [undefined, current] as const; + } + const next = new Map(current); + next.delete(threadId); + return [claim, next] as const; + }).pipe( + Effect.flatMap((released) => + released === undefined || released.phase !== "in-flight" + ? Effect.void + : Effect.all([ + registry.updateCursor(threadId, released.previousCursor, released.generation), + registry.setWakeCount(threadId, released.previousWakeCount, released.generation), + ]).pipe(Effect.asVoid), + ), + ); + + const dispatchUpdate = Effect.fn("PullRequestMonitor.dispatchUpdate")(function* ( + threadId: ThreadId, + blockersSummary: string, + headSha: string, + wakeCount: number, + ) { + const updatedAt = DateTime.formatIso(yield* DateTime.now); + yield* engine.dispatch({ + type: "thread.monitor.update", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId, + blockersSummary, + headSha, + wakeCount, + updatedAt, + }); + }); + + const end = Effect.fn("PullRequestMonitor.end")(function* ( + threadId: ThreadId, + reason: "ready" | "terminal" | "needs-attention" | "stopped", + blockersSummary: string, + expectedGeneration: number, + ) { + const current = yield* registry.get(threadId); + if (Option.isNone(current) || current.value.generation !== expectedGeneration) return; + // Dispatch before removing: if the end event fails to persist, the + // registration survives and the next poll retries the same terminal / + // ready / breaker verdict, instead of leaving the projection stuck on + // "monitoring" with nothing polling it. + const endedAt = DateTime.formatIso(yield* DateTime.now); + yield* engine.dispatch({ + type: "thread.monitor.end", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId, + reason, + blockersSummary, + endedAt, + }); + yield* registry.remove(threadId, expectedGeneration); + yield* mutateClaim(threadId, () => undefined); + }); + + const dispatchWake = Effect.fn("PullRequestMonitor.dispatchWake")(function* ( + registration: MonitorRegistry.MonitorRegistration, + snapshot: PullRequestMonitorSnapshot, + nextCursor: PullRequestMonitorCursor, + prompt: string, + ) { + const commandId = CommandId.make(yield* crypto.randomUUIDv4); + let claimed = false; + yield* Ref.update(claims, (current) => { + if (current.has(registration.threadId)) return current; + claimed = true; + return new Map(current).set(registration.threadId, { + generation: registration.generation, + commandId, + phase: "pending", + previousCursor: registration.cursor, + previousWakeCount: registration.wakeCount, + }); + }); + if (!claimed) return; + + const current = yield* registry.get(registration.threadId); + // Fresh terminal read at the send boundary (I1): the poll's snapshot is + // seconds stale by now, and a merge in that window must veto the wake — + // the cached `latestStates` alone cannot see it. + const freshState = yield* fetcher + .fetch({ cwd: registration.repoCwd, pullRequestNumber: registration.prNumber }) + .pipe( + Effect.map((fresh) => Option.some(fresh.state)), + Effect.orElseSucceed(() => Option.none()), + ); + if (Option.isNone(freshState)) { + yield* mutateClaim(registration.threadId, (claim) => + claim?.commandId === commandId ? undefined : claim, + ); + return; + } + const state = freshState.value; + const thread = yield* snapshots.getThreadDetailById(registration.threadId); + const validRegistration = + Option.isSome(current) && current.value.generation === registration.generation; + const busy = + Option.isNone(thread) || + thread.value.latestTurn?.state === "running" || + (thread.value.session?.activeTurnId !== null && + thread.value.session?.activeTurnId !== undefined) || + thread.value.session?.status === "starting" || + thread.value.session?.status === "running"; + const stillClaimed = + (yield* Ref.get(claims)).get(registration.threadId)?.commandId === commandId; + if (!validRegistration || state === "closed" || state === "merged" || busy || !stillClaimed) { + yield* mutateClaim(registration.threadId, (claim) => + claim?.commandId === commandId ? undefined : claim, + ); + return; + } + + const createdAt = DateTime.formatIso(yield* DateTime.now); + yield* engine + .dispatch({ + type: "thread.turn.start", + commandId, + threadId: registration.threadId, + message: { + messageId: MessageId.make(yield* crypto.randomUUIDv4), + role: "user", + text: prompt, + attachments: [], + }, + runtimeMode: thread.value.runtimeMode, + interactionMode: thread.value.interactionMode, + createdAt, + }) + .pipe( + Effect.tap(() => + // Stamp the claim with the command's own createdAt: a provider + // turn-start failure activity echoes exactly this timestamp + // (ProviderCommandReactor appends it with the event payload's + // createdAt), so equality means "our wake failed" while an older + // turn's delayed failure compares strictly older. A post-dispatch + // DateTime.now here would always exceed the echoed timestamp and + // make the release condition unsatisfiable. + mutateClaim(registration.threadId, (claim) => + claim?.commandId === commandId + ? { ...claim, phase: "in-flight", dispatchedAt: createdAt } + : claim, + ), + ), + Effect.tap(() => + registry.updateCursor(registration.threadId, nextCursor, registration.generation), + ), + Effect.tapError(() => + mutateClaim(registration.threadId, (claim) => + claim?.commandId === commandId ? undefined : claim, + ), + ), + ); + const wakeCount = yield* registry.incrementWake(registration.threadId, registration.generation); + yield* dispatchUpdate( + registration.threadId, + formatBlockersSummary(computeReadiness(snapshot)), + snapshot.headSha, + wakeCount, + ); + }); + + const pollRegistration = Effect.fn("PullRequestMonitor.pollRegistration")(function* ( + registration: MonitorRegistry.MonitorRegistration, + ) { + const fetchedSnapshot = yield* fetcher.fetch({ + cwd: registration.repoCwd, + pullRequestNumber: registration.prNumber, + }); + const snapshot = { + ...fetchedSnapshot, + monitoringStartedAt: registration.startedAt, + }; + const readiness = computeReadiness(snapshot); + const summary = formatBlockersSummary(readiness); + + if (snapshot.state !== "open") { + yield* end(registration.threadId, "terminal", summary, registration.generation); + return; + } + if (readiness.ready) { + yield* dispatchUpdate( + registration.threadId, + summary, + snapshot.headSha, + registration.wakeCount, + ); + yield* end(registration.threadId, "ready", summary, registration.generation); + return; + } + + const diff = diffPullRequestMonitorSnapshot(registration.cursor, snapshot); + yield* dispatchUpdate(registration.threadId, summary, snapshot.headSha, registration.wakeCount); + if (diff.actionableEvents.length === 0) return; + if (registration.wakeCount >= 10) { + yield* end(registration.threadId, "needs-attention", summary, registration.generation); + return; + } + yield* dispatchWake( + registration, + snapshot, + diff.nextCursor, + buildWakePrompt({ + prNumber: registration.prNumber, + wakeCount: registration.wakeCount + 1, + events: diff.actionableEvents, + snapshot, + readiness, + }), + ); + }); + + const reconcile = Effect.fn("PullRequestMonitor.reconcile")(function* () { + if (yield* Ref.get(reconciled)) return; + const shell = yield* snapshots.getShellSnapshot(); + yield* Effect.forEach( + shell.threads.filter((thread) => thread.monitor?.status === "monitoring"), + (thread) => + registry.get(thread.id).pipe( + Effect.flatMap((registration) => { + if (Option.isSome(registration)) return Effect.void; + return Effect.gen(function* () { + const endedAt = DateTime.formatIso(yield* DateTime.now); + yield* engine.dispatch({ + type: "thread.monitor.end", + commandId: CommandId.make(`monitor-boot-reconcile:${thread.id}:${endedAt}`), + threadId: thread.id, + reason: "session-ended", + blockersSummary: "", + endedAt, + }); + }); + }), + ), + { discard: true }, + ); + yield* Ref.set(reconciled, true); + }); + + const pollOnce = Ref.set(lastPollFailed, false).pipe( + Effect.andThen( + reconcile().pipe( + Effect.catch((error) => + Effect.logWarning("pull request monitor boot reconcile failed", { error }).pipe( + Effect.andThen(Ref.set(lastPollFailed, true)), + ), + ), + ), + ), + Effect.andThen(registry.listActive), + Effect.flatMap((registrations) => + Effect.forEach( + registrations, + (registration) => + pollRegistration(registration).pipe( + Effect.catchCause((cause) => + Effect.logWarning("pull request monitor poll failed", { + threadId: registration.threadId, + cause, + }).pipe(Effect.andThen(Ref.set(lastPollFailed, true))), + ), + ), + { concurrency: "unbounded", discard: true }, + ), + ), + ); + + const onEvent = (event: OrchestrationEvent) => { + const threadId = threadIdOf(event); + if (threadId === undefined) return Effect.void; + if (event.type === "thread.archived") { + return registry.get(threadId).pipe( + Effect.flatMap((registration) => + Option.isNone(registration) + ? Effect.void + : end(threadId, "stopped", "", registration.value.generation).pipe( + Effect.catch((error) => + Effect.logWarning( + "pull request monitor archive projection stuck; removing registration", + { + threadId, + error, + }, + ).pipe( + Effect.andThen(registry.remove(threadId, registration.value.generation)), + Effect.andThen( + mutateClaim(threadId, (claim) => + claim?.generation === registration.value.generation ? undefined : claim, + ), + ), + Effect.asVoid, + ), + ), + ), + ), + ); + } + if ( + event.type === "thread.deleted" || + event.type === "thread.settled" || + event.type === "thread.monitor-ended" + ) { + return registry + .remove(threadId) + .pipe(Effect.andThen(mutateClaim(threadId, () => undefined)), Effect.asVoid); + } + if (event.type === "thread.turn-start-requested") { + // A turn we didn't start won the thread. Drop our claim — and if our + // wake had already acked the cursor (in-flight), rewind it so the + // superseded events re-diff next poll instead of being lost (I5). + return releaseClaim(threadId, (claim) => claim.commandId !== event.commandId); + } + if ( + event.type === "thread.activity-appended" && + event.payload.activity.kind === "provider.turn.start.failed" + ) { + // engine.dispatch succeeded but the provider reactor failed the turn + // asynchronously — the acked events never reached an agent (I5). + return releaseClaim( + threadId, + (claim) => + claim.phase === "in-flight" && + claim.dispatchedAt !== undefined && + claim.dispatchedAt <= event.payload.activity.createdAt, + ); + } + if (event.type === "thread.session-set") { + const status = event.payload.session.status; + if (status !== "starting" && status !== "running") { + return mutateClaim(threadId, (claim) => (claim?.phase === "in-flight" ? undefined : claim)); + } + } + return Effect.void; + }; + + const start = Effect.fn("PullRequestMonitor.start")(function* () { + yield* Effect.forkDetach(Stream.runForEach(engine.streamDomainEvents, onEvent)); + yield* Effect.forkDetach( + Effect.forever( + pollOnce.pipe( + Effect.andThen(Ref.get(lastPollFailed)), + Effect.flatMap((failed) => Effect.sleep(failed ? MAX_BACKOFF : POLL_INTERVAL)), + ), + ), + ); + }); + + return PullRequestMonitor.of({ start, pollOnce, handleDomainEvent: onEvent }); +}); + +export const layer = Layer.effect(PullRequestMonitor, make); diff --git a/apps/server/src/monitor/monitorDiff.test.ts b/apps/server/src/monitor/monitorDiff.test.ts new file mode 100644 index 00000000000..a31d13770a8 --- /dev/null +++ b/apps/server/src/monitor/monitorDiff.test.ts @@ -0,0 +1,187 @@ +import { assert, describe, it } from "@effect/vitest"; + +import type { PullRequestMonitorSnapshot } from "../sourceControl/gitHubPullRequestMonitor.ts"; +import { cursorFromSnapshot, diffPullRequestMonitorSnapshot } from "./monitorDiff.ts"; + +const snapshot = ( + overrides: Partial = {}, +): PullRequestMonitorSnapshot => ({ + state: "open", + draft: false, + headSha: "head-1", + baseRefName: "main", + mergeability: "mergeable", + behindBaseBy: null, + requiredChecksKnown: true, + reviews: [], + reviewThreads: [], + issueComments: [], + checkRuns: [], + ...overrides, +}); + +const thread = (updatedAt = "2026-01-01T00:00:00Z", resolved = false) => ({ + id: "thread-1", + author: { login: "review-bot", type: "app" as const }, + latestCommentByViewer: false, + body: "Fix this", + path: "src/a.ts", + line: 3, + createdAt: "2026-01-01T00:00:00Z", + updatedAt, + resolved, +}); + +const check = (id: string, conclusion: "success" | "failure") => ({ + id, + name: "CI", + status: "completed" as const, + conclusion, + startedAt: "2026-01-01T00:00:00Z", + headSha: "head-1", +}); + +describe("pull request monitor diff", () => { + it("baselines every existing event", () => { + const initial = snapshot({ + reviewThreads: [thread()], + checkRuns: [check("run-1", "failure")], + }); + assert.deepStrictEqual( + diffPullRequestMonitorSnapshot(cursorFromSnapshot(initial), initial).actionableEvents, + [], + ); + }); + + it("detects a new comment and an edit by id plus updatedAt", () => { + const emptyCursor = cursorFromSnapshot(snapshot()); + const withComment = snapshot({ reviewThreads: [thread()] }); + const added = diffPullRequestMonitorSnapshot(emptyCursor, withComment); + assert.deepStrictEqual(added.actionableEvents, [ + { kind: "new-review-comment", threadId: "thread-1", edited: false }, + ]); + + const edited = diffPullRequestMonitorSnapshot( + added.nextCursor, + snapshot({ reviewThreads: [thread("2026-01-02T00:00:00Z")] }), + ); + assert.deepStrictEqual(edited.actionableEvents, [ + { kind: "new-review-comment", threadId: "thread-1", edited: true }, + ]); + }); + + it("does not wake for the viewer's reply, then wakes for a subsequent reviewer comment", () => { + const initial = snapshot({ reviewThreads: [thread()] }); + const selfReply = snapshot({ + reviewThreads: [ + { + ...thread("2026-01-02T00:00:00Z"), + author: { login: "claude", type: "user" as const }, + latestCommentByViewer: true, + }, + ], + }); + const ignored = diffPullRequestMonitorSnapshot(cursorFromSnapshot(initial), selfReply); + assert.deepStrictEqual(ignored.actionableEvents, []); + + const reviewerReply = snapshot({ + reviewThreads: [thread("2026-01-03T00:00:00Z")], + }); + assert.deepStrictEqual( + diffPullRequestMonitorSnapshot(ignored.nextCursor, reviewerReply).actionableEvents, + [{ kind: "new-review-comment", threadId: "thread-1", edited: true }], + ); + }); + + it("detects new and edited bot issue comments", () => { + const issueComment = { + id: "comment-1", + author: { login: "review-bot", type: "app" as const }, + body: "Please update this", + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + }; + const added = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(snapshot()), + snapshot({ issueComments: [issueComment] }), + ); + assert.deepStrictEqual(added.actionableEvents, [ + { kind: "new-review-comment", threadId: "comment-1", edited: false }, + ]); + const edited = diffPullRequestMonitorSnapshot( + added.nextCursor, + snapshot({ + issueComments: [{ ...issueComment, updatedAt: "2026-01-02T00:00:00Z" }], + }), + ); + assert.deepStrictEqual(edited.actionableEvents, [ + { kind: "new-review-comment", threadId: "comment-1", edited: true }, + ]); + }); + + it("records a resolved transition without waking", () => { + const initial = snapshot({ reviewThreads: [thread()] }); + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ reviewThreads: [thread("2026-01-02T00:00:00Z", true)] }), + ); + assert.deepStrictEqual(result.actionableEvents, []); + assert.strictEqual(result.nextCursor.threadVersions["thread-1"]?.resolved, true); + }); + + it("wakes when a resolved thread is reopened without an updatedAt change", () => { + const initial = snapshot({ reviewThreads: [thread(undefined, true)] }); + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ reviewThreads: [thread()] }), + ); + assert.deepStrictEqual(result.actionableEvents, [ + { kind: "new-review-comment", threadId: "thread-1", edited: true }, + ]); + }); + + it("wakes for a failed rerun with a new run id", () => { + const initial = snapshot({ checkRuns: [check("run-1", "failure")] }); + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ checkRuns: [check("run-2", "failure")] }), + ); + assert.deepStrictEqual(result.actionableEvents, [ + { kind: "check-failed", checkRunId: "run-2", checkName: "CI" }, + ]); + }); + + it("wakes for a new failure after a seen success", () => { + const initial = snapshot({ checkRuns: [check("run-1", "success")] }); + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ checkRuns: [check("run-2", "failure")] }), + ); + assert.deepStrictEqual(result.actionableEvents, [ + { kind: "check-failed", checkRunId: "run-2", checkName: "CI" }, + ]); + }); + + it("wakes for a same-name failure on a new head", () => { + const initial = snapshot({ checkRuns: [check("run-1", "failure")] }); + const nextCheck = { ...check("run-2", "failure"), headSha: "head-2" }; + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ headSha: "head-2", checkRuns: [nextCheck] }), + ); + assert.deepStrictEqual(result.actionableEvents, [ + { kind: "check-failed", checkRunId: "run-2", checkName: "CI" }, + ]); + }); + + it("tracks concurrent same-name runs independently", () => { + const initial = snapshot({ checkRuns: [check("run-1", "success")] }); + const result = diffPullRequestMonitorSnapshot( + cursorFromSnapshot(initial), + snapshot({ checkRuns: [check("run-1", "success"), check("run-2", "failure")] }), + ); + assert.deepStrictEqual(result.actionableEvents, [ + { kind: "check-failed", checkRunId: "run-2", checkName: "CI" }, + ]); + }); +}); diff --git a/apps/server/src/monitor/monitorDiff.ts b/apps/server/src/monitor/monitorDiff.ts new file mode 100644 index 00000000000..7aaeda8186e --- /dev/null +++ b/apps/server/src/monitor/monitorDiff.ts @@ -0,0 +1,120 @@ +import type { + PullRequestMonitorCheckRun, + PullRequestMonitorSnapshot, +} from "../sourceControl/gitHubPullRequestMonitor.ts"; + +export interface PullRequestMonitorCursor { + readonly headSha: string; + readonly reviewStates: Readonly>; + readonly threadVersions: Readonly< + Record + >; + readonly issueCommentVersions: Readonly>; + readonly checkRuns: Readonly< + Record + >; + readonly behindBase: boolean; +} + +export type PullRequestMonitorActionableEvent = + | { readonly kind: "new-review-comment"; readonly threadId: string; readonly edited: boolean } + | { readonly kind: "changes-requested-review"; readonly reviewId: string } + | { readonly kind: "check-failed"; readonly checkRunId: string; readonly checkName: string } + | { readonly kind: "behind-base" }; + +const checkOutcome = (check: PullRequestMonitorCheckRun): "success" | "failure" | "pending" => { + if (check.status !== "completed") return "pending"; + if ( + check.conclusion === "success" || + check.conclusion === "neutral" || + check.conclusion === "skipped" + ) { + return "success"; + } + return "failure"; +}; + +export function cursorFromSnapshot(snapshot: PullRequestMonitorSnapshot): PullRequestMonitorCursor { + return { + headSha: snapshot.headSha, + reviewStates: Object.fromEntries(snapshot.reviews.map((review) => [review.id, review.state])), + threadVersions: Object.fromEntries( + snapshot.reviewThreads.map((thread) => [ + thread.id, + { updatedAt: thread.updatedAt, resolved: thread.resolved }, + ]), + ), + issueCommentVersions: Object.fromEntries( + snapshot.issueComments.map((comment) => [comment.id, comment.updatedAt]), + ), + checkRuns: Object.fromEntries( + snapshot.checkRuns + .filter((check) => check.headSha === snapshot.headSha) + .map((check) => [ + `${check.headSha}::${check.name}::${check.id}`, + { runId: check.id, outcome: checkOutcome(check) }, + ]), + ), + behindBase: (snapshot.behindBaseBy ?? 0) > 0, + }; +} + +export function diffPullRequestMonitorSnapshot( + previousCursor: PullRequestMonitorCursor, + snapshot: PullRequestMonitorSnapshot, +): { + readonly actionableEvents: ReadonlyArray; + readonly nextCursor: PullRequestMonitorCursor; +} { + const actionableEvents: PullRequestMonitorActionableEvent[] = []; + for (const thread of snapshot.reviewThreads) { + const previous = previousCursor.threadVersions[thread.id]; + if ( + !thread.latestCommentByViewer && + !thread.resolved && + (!previous || previous.resolved || previous.updatedAt !== thread.updatedAt) + ) { + actionableEvents.push({ + kind: "new-review-comment", + threadId: thread.id, + edited: previous !== undefined, + }); + } + } + for (const comment of snapshot.issueComments) { + const previousUpdatedAt = previousCursor.issueCommentVersions[comment.id]; + if (!previousUpdatedAt || previousUpdatedAt !== comment.updatedAt) { + actionableEvents.push({ + kind: "new-review-comment", + threadId: comment.id, + edited: previousUpdatedAt !== undefined, + }); + } + } + for (const review of snapshot.reviews) { + if ( + review.commitSha === snapshot.headSha && + review.state === "changes-requested" && + previousCursor.reviewStates[review.id] !== "changes-requested" + ) { + actionableEvents.push({ kind: "changes-requested-review", reviewId: review.id }); + } + } + for (const check of snapshot.checkRuns.filter((item) => item.headSha === snapshot.headSha)) { + const outcome = checkOutcome(check); + const previous = previousCursor.checkRuns[`${check.headSha}::${check.name}::${check.id}`]; + // Each concrete run is acknowledged independently, including concurrent + // same-name runs on one head and reruns/new heads with new run ids. + if (outcome === "failure" && previous?.outcome !== "failure") { + actionableEvents.push({ + kind: "check-failed", + checkRunId: check.id, + checkName: check.name, + }); + } + } + if ((snapshot.behindBaseBy ?? 0) > 0 && !previousCursor.behindBase) { + actionableEvents.push({ kind: "behind-base" }); + } + return { actionableEvents, nextCursor: cursorFromSnapshot(snapshot) }; +} diff --git a/apps/server/src/monitor/readiness.test.ts b/apps/server/src/monitor/readiness.test.ts new file mode 100644 index 00000000000..0c25491a5b8 --- /dev/null +++ b/apps/server/src/monitor/readiness.test.ts @@ -0,0 +1,112 @@ +import { assert, describe, it } from "@effect/vitest"; + +import type { PullRequestMonitorSnapshot } from "../sourceControl/gitHubPullRequestMonitor.ts"; +import { computeReadiness } from "./readiness.ts"; + +const snapshot = ( + overrides: Partial = {}, +): PullRequestMonitorSnapshot => ({ + state: "open", + draft: false, + headSha: "head-2", + baseRefName: "main", + mergeability: "mergeable", + behindBaseBy: null, + requiredChecksKnown: true, + reviews: [], + reviewThreads: [], + issueComments: [], + checkRuns: [ + { + id: "run-1", + name: "CI", + status: "completed", + conclusion: "success", + startedAt: null, + headSha: "head-2", + }, + ], + ...overrides, +}); + +describe("pull request monitor readiness", () => { + it("keeps blocking on a changes-requested review even after a push", () => { + // A stale change request must hold the gate until the bot re-reviews; + // otherwise every fix push would flash green before re-review lands. + const result = computeReadiness( + snapshot({ + reviews: [ + { + id: "review-1", + author: { login: "review-bot", type: "app" }, + state: "changes-requested", + submittedAt: "2026-01-01T00:00:00Z", + commitSha: "head-1", + }, + ], + }), + ); + assert.strictEqual(result.ready, false); + assert.deepStrictEqual(result.blockers, [ + { kind: "changes-requested", reviewer: "review-bot" }, + ]); + }); + + it("blocks on a human changes-requested review", () => { + const result = computeReadiness( + snapshot({ + reviews: [ + { + id: "review-human", + author: { login: "human-reviewer", type: "user" }, + state: "changes-requested", + submittedAt: "2026-01-01T00:00:00Z", + commitSha: "head-2", + }, + ], + }), + ); + assert.deepStrictEqual(result.blockers, [ + { kind: "changes-requested", reviewer: "human-reviewer" }, + ]); + }); + + it("ignores stale approvals after a push", () => { + const result = computeReadiness( + snapshot({ + reviews: [ + { + id: "review-1", + author: { login: "review-bot", type: "app" }, + state: "approved", + submittedAt: "2026-01-01T00:00:00Z", + commitSha: "head-1", + }, + ], + }), + ); + // A stale approval neither blocks nor counts as fresh green evidence. + assert.strictEqual(result.ready, true); + }); + + it("does not report ready when no check runs exist", () => { + const result = computeReadiness(snapshot({ checkRuns: [] })); + assert.strictEqual(result.ready, false); + assert.strictEqual(result.label, "no-known-blockers"); + assert.deepStrictEqual(result.blockers, [{ kind: "checks-missing" }]); + }); + + it("returns terminal blockers for merged and closed snapshots", () => { + for (const state of ["merged", "closed"] as const) { + const result = computeReadiness(snapshot({ state })); + assert.strictEqual(result.ready, false); + assert.deepStrictEqual(result.blockers[0], { kind: "terminal", state }); + } + }); + + it("uses the honest fallback label when required-check evidence is unavailable", () => { + const result = computeReadiness(snapshot({ requiredChecksKnown: false })); + assert.strictEqual(result.ready, true); + assert.strictEqual(result.label, "no-known-blockers"); + }); +}); diff --git a/apps/server/src/monitor/readiness.ts b/apps/server/src/monitor/readiness.ts new file mode 100644 index 00000000000..3daafe89e2d --- /dev/null +++ b/apps/server/src/monitor/readiness.ts @@ -0,0 +1,64 @@ +import type { PullRequestMonitorSnapshot } from "../sourceControl/gitHubPullRequestMonitor.ts"; + +export type PullRequestMonitorBlocker = + | { readonly kind: "terminal"; readonly state: "closed" | "merged" } + | { readonly kind: "draft" } + | { readonly kind: "mergeability"; readonly state: "conflicting" | "unknown" } + | { readonly kind: "checks-missing" } + | { readonly kind: "check-pending"; readonly checkName: string } + | { readonly kind: "check-failed"; readonly checkName: string } + | { readonly kind: "changes-requested"; readonly reviewer: string } + | { readonly kind: "unresolved-thread"; readonly threadId: string }; + +export interface PullRequestMonitorReadiness { + readonly ready: boolean; + readonly label: "ready-to-merge" | "no-known-blockers"; + readonly blockers: ReadonlyArray; +} + +export function computeReadiness( + snapshot: PullRequestMonitorSnapshot, +): PullRequestMonitorReadiness { + const blockers: PullRequestMonitorBlocker[] = []; + if (snapshot.state !== "open") blockers.push({ kind: "terminal", state: snapshot.state }); + if (snapshot.draft) blockers.push({ kind: "draft" }); + if (snapshot.mergeability !== "mergeable") { + blockers.push({ kind: "mergeability", state: snapshot.mergeability }); + } + const currentChecks = snapshot.checkRuns.filter((check) => check.headSha === snapshot.headSha); + if (currentChecks.length === 0) blockers.push({ kind: "checks-missing" }); + for (const check of currentChecks) { + if (check.status !== "completed") { + blockers.push({ kind: "check-pending", checkName: check.name }); + } else if ( + check.conclusion !== "success" && + check.conclusion !== "neutral" && + check.conclusion !== "skipped" + ) { + blockers.push({ kind: "check-failed", checkName: check.name }); + } + } + for (const review of snapshot.reviews) { + // Stale approvals must not count toward green, but a changes-requested + // review blocks regardless of which commit it reviewed — GitHub keeps it + // active until dismissed or superseded by a re-review. + if (review.state === "changes-requested") { + blockers.push({ kind: "changes-requested", reviewer: review.author.login }); + } + } + for (const thread of snapshot.reviewThreads) { + if ( + !thread.resolved && + (!snapshot.monitoringStartedAt || thread.createdAt >= snapshot.monitoringStartedAt) + ) { + blockers.push({ kind: "unresolved-thread", threadId: thread.id }); + } + } + + const evidenceSupportsReadyLabel = snapshot.requiredChecksKnown && currentChecks.length > 0; + return { + ready: blockers.length === 0, + label: evidenceSupportsReadyLabel ? "ready-to-merge" : "no-known-blockers", + blockers, + }; +} diff --git a/apps/server/src/monitor/wakePrompt.ts b/apps/server/src/monitor/wakePrompt.ts new file mode 100644 index 00000000000..5d72fca76e5 --- /dev/null +++ b/apps/server/src/monitor/wakePrompt.ts @@ -0,0 +1,87 @@ +import type { PullRequestMonitorSnapshot } from "../sourceControl/gitHubPullRequestMonitor.ts"; +import type { PullRequestMonitorActionableEvent } from "./monitorDiff.ts"; +import type { PullRequestMonitorReadiness } from "./readiness.ts"; + +const excerpt = (body: string) => body.replace(/\s+/g, " ").trim().slice(0, 280); + +export function formatBlockersSummary(readiness: PullRequestMonitorReadiness): string { + if (readiness.blockers.length === 0) + return readiness.label === "ready-to-merge" ? "Ready to merge" : "No known blockers"; + return readiness.blockers + .map((blocker) => { + switch (blocker.kind) { + case "terminal": + return `PR is ${blocker.state}`; + case "draft": + return "PR is a draft"; + case "mergeability": + return `Mergeability: ${blocker.state}`; + case "checks-missing": + return "No check results are available"; + case "check-pending": + return `${blocker.checkName}: pending`; + case "check-failed": + return `${blocker.checkName}: failed`; + case "changes-requested": + return `${blocker.reviewer}: changes requested`; + case "unresolved-thread": + return `Review thread ${blocker.threadId}: unresolved`; + } + }) + .join("\n"); +} + +function formatEvent( + event: PullRequestMonitorActionableEvent, + snapshot: PullRequestMonitorSnapshot, +): string { + switch (event.kind) { + case "new-review-comment": { + const thread = snapshot.reviewThreads.find((item) => item.id === event.threadId); + const comment = snapshot.issueComments.find((item) => item.id === event.threadId); + if (thread) { + const location = + thread.path === null + ? "" + : `, ${thread.path}${thread.line === null ? "" : `:${thread.line}`}`; + return `- Comment from ${thread.author.login}${location}: ${excerpt(thread.body)}`; + } + return comment + ? `- Comment from ${comment.author.login}: ${excerpt(comment.body)}` + : `- ${event.edited ? "Updated" : "New"} review comment (${event.threadId})`; + } + case "changes-requested-review": { + const review = snapshot.reviews.find((item) => item.id === event.reviewId); + return `- ${review?.author.login ?? "Reviewer"} requested changes`; + } + case "check-failed": { + const check = snapshot.checkRuns.find((item) => item.id === event.checkRunId); + return `- Check ${event.checkName}: ${check?.conclusion ?? "failure"}`; + } + case "behind-base": + return `- PR is behind ${snapshot.baseRefName}${snapshot.behindBaseBy === null ? "" : ` by ${snapshot.behindBaseBy} commit(s)`}`; + } +} + +export function buildWakePrompt(input: { + readonly prNumber: number; + readonly wakeCount: number; + readonly events: ReadonlyArray; + readonly snapshot: PullRequestMonitorSnapshot; + readonly readiness: PullRequestMonitorReadiness; +}): string { + return `New activity on PR #${input.prNumber} (monitoring, wake ${input.wakeCount}/10). + +${input.events.map((event) => formatEvent(event, input.snapshot)).join("\n")} + +Status: ${formatBlockersSummary(input.readiness)} +Head: ${input.snapshot.headSha} (approvals/checks are evaluated against this commit) + +Policy: +- Verify bot claims against the source before acting. +- Fix legitimate findings and push. +- Dismiss false positives with a brief reply — never silently ignore or comply. +- For CI failures: compare against ${input.snapshot.baseRefName}; re-run suspected flakes; if the same real failure repeats, ask the user via a question rather than guessing. +- Rebase if the PR is behind ${input.snapshot.baseRefName}. +- If the goal has become impossible, say so and ask the user.`; +} diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1f24a4a0200..c6eccc6590b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -611,6 +611,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -713,6 +714,70 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.monitor-started": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + monitor: { + prNumber: event.payload.prNumber, + status: "monitoring", + blockersSummary: event.payload.blockersSummary, + headSha: event.payload.headSha, + wakeCount: event.payload.wakeCount, + startedAt: event.payload.startedAt, + endedAt: null, + endedReason: null, + }, + updatedAt: event.payload.startedAt, + }); + return; + } + + case "thread.monitor-snapshot-updated": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow) || existingRow.value.monitor === null) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + monitor: { + ...existingRow.value.monitor, + blockersSummary: event.payload.blockersSummary, + headSha: event.payload.headSha, + wakeCount: event.payload.wakeCount, + }, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.monitor-ended": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow) || existingRow.value.monitor === null) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + monitor: { + ...existingRow.value.monitor, + status: + event.payload.reason === "ready" + ? "ready" + : event.payload.reason === "needs-attention" + ? "needs-attention" + : "stopped", + blockersSummary: event.payload.blockersSummary, + endedAt: event.payload.endedAt, + endedReason: event.payload.reason, + }, + updatedAt: event.payload.endedAt, + }); + return; + } + case "thread.meta-updated": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index d4a24a209ad..161bf7ee5ef 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -312,6 +312,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, deletedAt: null, messages: [ { @@ -426,6 +427,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, session: { threadId: ThreadId.make("thread-1"), status: "running", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 3d05bef4bdf..b50be3eafba 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -10,6 +10,7 @@ import { OrchestrationShellSnapshot, OrchestrationThread, OrchestrationThreadDetailSnapshot, + OrchestrationThreadMonitor, ProjectScript, TurnId, type OrchestrationCheckpointSummary, @@ -78,6 +79,7 @@ const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan; const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), + monitor: Schema.NullOr(Schema.fromJsonString(OrchestrationThreadMonitor)), }), ); const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( @@ -338,6 +340,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -370,6 +373,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -404,6 +408,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -770,6 +775,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1206,6 +1212,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: row.settledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + monitor: row.monitor, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -1408,6 +1415,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: row.settledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + monitor: row.monitor, deletedAt: row.deletedAt, messages: [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -1541,6 +1549,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: row.settledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + monitor: row.monitor, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -1679,6 +1688,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: row.settledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + monitor: row.monitor, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -1923,6 +1933,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: threadRow.value.settledAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, + monitor: threadRow.value.monitor, session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, latestUserMessageAt: threadRow.value.latestUserMessageAt, hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, @@ -2021,6 +2032,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { settledAt: threadRow.value.settledAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, + monitor: threadRow.value.monitor, deletedAt: null, messages: messageRows.map((row) => { const message = { diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 3b558d24739..1c5e19c4e09 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -13,6 +13,9 @@ import { ThreadUnsettledPayload as ContractsThreadUnsettledPayloadSchema, ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, + ThreadMonitorStartedPayload as ContractsThreadMonitorStartedPayloadSchema, + ThreadMonitorSnapshotUpdatedPayload as ContractsThreadMonitorSnapshotUpdatedPayloadSchema, + ThreadMonitorEndedPayload as ContractsThreadMonitorEndedPayloadSchema, ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema, ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema, ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema, @@ -42,6 +45,10 @@ export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; +export const ThreadMonitorStartedPayload = ContractsThreadMonitorStartedPayloadSchema; +export const ThreadMonitorSnapshotUpdatedPayload = + ContractsThreadMonitorSnapshotUpdatedPayloadSchema; +export const ThreadMonitorEndedPayload = ContractsThreadMonitorEndedPayloadSchema; export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema; export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema; diff --git a/apps/server/src/orchestration/decider.monitor.test.ts b/apps/server/src/orchestration/decider.monitor.test.ts new file mode 100644 index 00000000000..a95a9d3ca34 --- /dev/null +++ b/apps/server/src/orchestration/decider.monitor.test.ts @@ -0,0 +1,179 @@ +import { + CommandId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationReadModel, + type OrchestrationThread, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-07-23T12:00:00.000Z"; + +const makeReadModel = (monitor: OrchestrationThread["monitor"] = null): OrchestrationReadModel => ({ + snapshotSequence: 0, + projects: [], + threads: [ + { + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { instanceId: ProviderInstanceId.make("claude"), model: "claude-opus" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "feature", + worktreePath: "/repo", + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + monitor, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: NOW, +}); + +const activeMonitor: NonNullable = { + prNumber: 42, + status: "monitoring", + blockersSummary: "waiting", + headSha: "abc", + wakeCount: 4, + startedAt: NOW, + endedAt: null, + endedReason: null, +}; + +const commandId = (value: string) => CommandId.make(value); +const threadId = ThreadId.make("thread-1"); + +it.layer(NodeServices.layer)("monitor decider", (it) => { + it.effect("starts and idempotently re-emits the same PR", () => + Effect.gen(function* () { + const start = { + type: "thread.monitor.start" as const, + commandId: commandId("start"), + threadId, + prNumber: 42, + blockersSummary: "", + headSha: "abc", + createdAt: NOW, + }; + const event = yield* decideOrchestrationCommand({ + command: start, + readModel: makeReadModel(), + }); + expect("type" in event ? event.type : event[0]?.type).toBe("thread.monitor-started"); + const repeated = yield* decideOrchestrationCommand({ + command: { ...start, commandId: commandId("again"), headSha: "new" }, + readModel: makeReadModel(activeMonitor), + }); + if ("type" in repeated && repeated.type === "thread.monitor-started") { + const payload = repeated.payload as { + readonly headSha: string; + readonly wakeCount: number; + readonly startedAt: string; + }; + expect(payload.headSha).toBe("abc"); + expect(payload.wakeCount).toBe(4); + expect(payload.startedAt).toBe(NOW); + } + }), + ); + + it.effect("rejects a different PR and updates/ends only an active monitor", () => + Effect.gen(function* () { + const different = yield* decideOrchestrationCommand({ + command: { + type: "thread.monitor.start", + commandId: commandId("different"), + threadId, + prNumber: 43, + blockersSummary: "", + headSha: "def", + createdAt: NOW, + }, + readModel: makeReadModel(activeMonitor), + }).pipe(Effect.flip); + expect(different._tag).toBe("OrchestrationCommandInvariantError"); + + const update = yield* decideOrchestrationCommand({ + command: { + type: "thread.monitor.update", + commandId: commandId("update"), + threadId, + blockersSummary: "CI", + headSha: "def", + wakeCount: 1, + updatedAt: NOW, + }, + readModel: makeReadModel(activeMonitor), + }); + expect("type" in update ? update.type : update[0]?.type).toBe( + "thread.monitor-snapshot-updated", + ); + const end = yield* decideOrchestrationCommand({ + command: { + type: "thread.monitor.end", + commandId: commandId("end"), + threadId, + reason: "ready", + blockersSummary: "", + endedAt: NOW, + }, + readModel: makeReadModel(activeMonitor), + }); + expect("type" in end ? end.type : end[0]?.type).toBe("thread.monitor-ended"); + }), + ); + + it.effect("rejects update/end without an active monitor and settle ends monitoring", () => + Effect.gen(function* () { + for (const command of [ + { + type: "thread.monitor.update" as const, + commandId: commandId("bad-update"), + threadId, + blockersSummary: "", + headSha: "abc", + wakeCount: 0, + updatedAt: NOW, + }, + { + type: "thread.monitor.end" as const, + commandId: commandId("bad-end"), + threadId, + reason: "stopped" as const, + blockersSummary: "", + endedAt: NOW, + }, + ]) { + const error = yield* decideOrchestrationCommand({ + command, + readModel: makeReadModel(), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + } + const settled = yield* decideOrchestrationCommand({ + command: { type: "thread.settle", commandId: commandId("settle"), threadId }, + readModel: makeReadModel(activeMonitor), + }); + expect(Array.isArray(settled) ? settled.map((event) => event.type) : []).toEqual([ + "thread.monitor-ended", + "thread.settled", + ]); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 73f1cbf9127..0c766214927 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -44,6 +44,7 @@ function makeReadModel( archivedAt, settledOverride, settledAt: settledOverride === "settled" ? SETTLED_AT : null, + monitor: null, deletedAt: null, messages, proposedPlans: [], diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 100369ae6e3..41eb852cc58 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -487,7 +487,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" // settledAt: the engine rejects zero-event commands, and bulk-settle / // double-click must stay silent no-ops rather than surface errors. const alreadySettled = thread.settledOverride === "settled" && thread.settledAt !== null; - return { + const settledEvent: PlannedOrchestrationEvent = { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, @@ -504,6 +504,143 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" updatedAt: alreadySettled ? thread.updatedAt : occurredAt, }, }; + if (thread.monitor?.status !== "monitoring") return settledEvent; + return [ + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.monitor-ended", + payload: { + threadId: command.threadId, + reason: "stopped", + blockersSummary: thread.monitor.blockersSummary, + endedAt: occurredAt, + }, + }, + settledEvent, + ]; + } + + case "thread.monitor.start": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + if (thread.monitor?.status === "monitoring" && thread.monitor.prNumber !== command.prNumber) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} is already monitoring PR #${thread.monitor.prNumber}`, + }); + } + const monitorStartedEvent: Omit = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.monitor-started", + payload: { + threadId: command.threadId, + prNumber: command.prNumber, + blockersSummary: + thread.monitor?.status === "monitoring" + ? thread.monitor.blockersSummary + : command.blockersSummary, + headSha: + thread.monitor?.status === "monitoring" ? thread.monitor.headSha : command.headSha, + wakeCount: thread.monitor?.status === "monitoring" ? thread.monitor.wakeCount : 0, + startedAt: + thread.monitor?.status === "monitoring" ? thread.monitor.startedAt : command.createdAt, + }, + }; + // Monitoring holds the thread unsettled (I2). Like turn.start, starting + // a monitor is real activity: it must clear any settled override so the + // projection can't carry "settled" and "monitoring" simultaneously. + if (thread.settledOverride === null) { + return monitorStartedEvent; + } + return [ + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.unsettled", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }, + monitorStartedEvent, + ]; + } + + case "thread.monitor.update": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + if (thread.monitor?.status !== "monitoring") { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} has no active monitor`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.updatedAt, + commandId: command.commandId, + })), + type: "thread.monitor-snapshot-updated", + payload: { + threadId: command.threadId, + blockersSummary: command.blockersSummary, + headSha: command.headSha, + wakeCount: command.wakeCount, + updatedAt: command.updatedAt, + }, + }; + } + + case "thread.monitor.end": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + if (thread.monitor?.status !== "monitoring") { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} has no active monitor`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.endedAt, + commandId: command.commandId, + })), + type: "thread.monitor-ended", + payload: { + threadId: command.threadId, + reason: command.reason, + blockersSummary: command.blockersSummary, + endedAt: command.endedAt, + }, + }; } case "thread.unsettle": { diff --git a/apps/server/src/orchestration/projector.monitor.test.ts b/apps/server/src/orchestration/projector.monitor.test.ts new file mode 100644 index 00000000000..038a74bfc3e --- /dev/null +++ b/apps/server/src/orchestration/projector.monitor.test.ts @@ -0,0 +1,83 @@ +import { + CommandId, + EventId, + ProjectId, + ThreadId, + type OrchestrationEvent, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const NOW = "2026-07-23T12:00:00.000Z"; +const event = (sequence: number, type: OrchestrationEvent["type"], payload: unknown) => + ({ + sequence, + eventId: EventId.make(`event-${sequence}`), + type, + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + occurredAt: NOW, + commandId: CommandId.make(`command-${sequence}`), + causationEventId: null, + correlationId: null, + metadata: {}, + payload, + }) as OrchestrationEvent; + +it.effect("projects the monitor lifecycle without changing settledOverride", () => + Effect.gen(function* () { + let model = yield* projectEvent( + createEmptyReadModel(NOW), + event(1, "thread.created", { + threadId: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { provider: "codex", model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: NOW, + updatedAt: NOW, + }), + ); + model = yield* projectEvent( + model, + event(2, "thread.monitor-started", { + threadId: ThreadId.make("thread-1"), + prNumber: 42, + blockersSummary: "draft", + headSha: "abc", + wakeCount: 0, + startedAt: NOW, + }), + ); + expect(model.threads[0]?.monitor?.status).toBe("monitoring"); + expect(model.threads[0]?.settledOverride).toBeNull(); + model = yield* projectEvent( + model, + event(3, "thread.monitor-snapshot-updated", { + threadId: ThreadId.make("thread-1"), + blockersSummary: "CI", + headSha: "def", + wakeCount: 2, + updatedAt: NOW, + }), + ); + expect(model.threads[0]?.monitor?.wakeCount).toBe(2); + model = yield* projectEvent( + model, + event(4, "thread.monitor-ended", { + threadId: ThreadId.make("thread-1"), + reason: "ready", + blockersSummary: "", + endedAt: NOW, + }), + ); + expect(model.threads[0]?.monitor?.status).toBe("ready"); + expect(model.threads[0]?.monitor?.endedReason).toBe("ready"); + expect(model.threads[0]?.settledOverride).toBeNull(); + }), +); diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 9c07a312023..a8692530724 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -93,6 +93,7 @@ describe("orchestration projector", () => { settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, deletedAt: null, messages: [], proposedPlans: [], diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 0504cb36f9a..9e2f8148c34 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -20,6 +20,9 @@ import { ThreadDeletedPayload, ThreadInteractionModeSetPayload, ThreadMetaUpdatedPayload, + ThreadMonitorEndedPayload, + ThreadMonitorSnapshotUpdatedPayload, + ThreadMonitorStartedPayload, ThreadProposedPlanUpsertedPayload, ThreadRuntimeModeSetPayload, ThreadSettledPayload, @@ -294,6 +297,7 @@ export function projectEvent( settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, deletedAt: null, messages: [], activities: [], @@ -393,6 +397,79 @@ export function projectEvent( })), ); + case "thread.monitor-started": + return decodeForEvent(ThreadMonitorStartedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + monitor: { + prNumber: payload.prNumber, + status: "monitoring", + blockersSummary: payload.blockersSummary, + headSha: payload.headSha, + wakeCount: payload.wakeCount, + startedAt: payload.startedAt, + endedAt: null, + endedReason: null, + }, + updatedAt: payload.startedAt, + }), + })), + ); + + case "thread.monitor-snapshot-updated": + return decodeForEvent( + ThreadMonitorSnapshotUpdatedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: nextBase.threads.map((thread) => + thread.id === payload.threadId && thread.monitor != null + ? { + ...thread, + monitor: { + ...thread.monitor, + blockersSummary: payload.blockersSummary, + headSha: payload.headSha, + wakeCount: payload.wakeCount, + }, + updatedAt: payload.updatedAt, + } + : thread, + ), + })), + ); + + case "thread.monitor-ended": + return decodeForEvent(ThreadMonitorEndedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: nextBase.threads.map((thread) => + thread.id === payload.threadId && thread.monitor != null + ? { + ...thread, + monitor: { + ...thread.monitor, + status: + payload.reason === "ready" + ? "ready" + : payload.reason === "needs-attention" + ? "needs-attention" + : "stopped", + blockersSummary: payload.blockersSummary, + endedAt: payload.endedAt, + endedReason: payload.reason, + }, + updatedAt: payload.endedAt, + } + : thread, + ), + })), + ); + case "thread.meta-updated": return decodeForEvent(ThreadMetaUpdatedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index 4763f565653..6ef1f418a22 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -95,6 +95,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { settledAt: null, snoozedUntil: null, snoozedAt: null, + monitor: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, @@ -157,6 +158,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { settledAt: "2026-03-25T00:00:00.000Z", snoozedUntil: "2026-03-26T09:00:00.000Z", snoozedAt: "2026-03-25T00:00:00.000Z", + monitor: null, latestUserMessageAt: null, pendingApprovalCount: 0, pendingUserInputCount: 0, diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 7e86d49eac3..4b8c8d5964c 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -14,11 +14,12 @@ import { ProjectionThreadRepository, type ProjectionThreadRepositoryShape, } from "../Services/ProjectionThreads.ts"; -import { ModelSelection } from "@t3tools/contracts"; +import { ModelSelection, OrchestrationThreadMonitor } from "@t3tools/contracts"; const ProjectionThreadDbRow = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), + monitor: Schema.NullOr(Schema.fromJsonString(OrchestrationThreadMonitor)), }), ); type ProjectionThreadDbRow = typeof ProjectionThreadDbRow.Type; @@ -47,6 +48,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { settled_at, snoozed_until, snoozed_at, + monitor_json, latest_user_message_at, pending_approval_count, pending_user_input_count, @@ -70,6 +72,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.settledAt}, ${row.snoozedUntil}, ${row.snoozedAt}, + ${row.monitor === null ? null : JSON.stringify(row.monitor)}, ${row.latestUserMessageAt}, ${row.pendingApprovalCount}, ${row.pendingUserInputCount}, @@ -93,6 +96,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { settled_at = excluded.settled_at, snoozed_until = excluded.snoozed_until, snoozed_at = excluded.snoozed_at, + monitor_json = excluded.monitor_json, latest_user_message_at = excluded.latest_user_message_at, pending_approval_count = excluded.pending_approval_count, pending_user_input_count = excluded.pending_user_input_count, @@ -123,6 +127,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -155,6 +160,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { settled_at AS "settledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + monitor_json AS "monitor", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index d25895671a9..75b0d5bcb63 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -47,6 +47,7 @@ import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts"; import Migration0034 from "./Migrations/034_ProjectionThreadsSnoozed.ts"; +import Migration0035 from "./Migrations/035_ProjectionThreadsMonitor.ts"; /** * Migration loader with all migrations defined inline. @@ -93,6 +94,7 @@ export const migrationEntries = [ [32, "AuthPairingProofKeyThumbprint", Migration0032], [33, "ProjectionThreadsSettled", Migration0033], [34, "ProjectionThreadsSnoozed", Migration0034], + [35, "ProjectionThreadsMonitor", Migration0035], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/035_ProjectionThreadsMonitor.ts b/apps/server/src/persistence/Migrations/035_ProjectionThreadsMonitor.ts new file mode 100644 index 00000000000..6f721c8bd4d --- /dev/null +++ b/apps/server/src/persistence/Migrations/035_ProjectionThreadsMonitor.ts @@ -0,0 +1,15 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + if (!columns.some((column) => column.name === "monitor_json")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN monitor_json TEXT + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 056425ae886..997ffbac0b4 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -9,6 +9,7 @@ import { IsoDateTime, ModelSelection, + OrchestrationThreadMonitor, NonNegativeInt, ProjectId, ProviderInteractionMode, @@ -40,6 +41,7 @@ export const ProjectionThread = Schema.Struct({ settledAt: Schema.NullOr(IsoDateTime), snoozedUntil: Schema.NullOr(IsoDateTime), snoozedAt: Schema.NullOr(IsoDateTime), + monitor: Schema.NullOr(OrchestrationThreadMonitor), latestUserMessageAt: Schema.NullOr(IsoDateTime), pendingApprovalCount: NonNegativeInt, pendingUserInputCount: NonNegativeInt, diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2eaaeb8ce3c..e3698322293 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -55,6 +55,7 @@ import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts"; +import * as MonitorRegistry from "../../monitor/MonitorRegistry.ts"; const isModelSelection = Schema.is(ModelSelection); /** @@ -223,7 +224,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ); const clearMcpSession = (threadId: ThreadId) => - McpSessionRegistry.revokeActiveMcpThread(threadId).pipe( + MonitorRegistry.endActiveMonitorForSession(threadId).pipe( + Effect.andThen(McpSessionRegistry.revokeActiveMcpThread(threadId)), Effect.tap(() => Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId))), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 66b9823afb3..96c91969d9f 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -36,6 +36,8 @@ import { ProviderInstanceRegistryHydrationLive } from "./provider/Layers/Provide import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; +import * as MonitorRegistry from "./monitor/MonitorRegistry.ts"; +import * as PullRequestMonitor from "./monitor/PullRequestMonitor.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; @@ -158,6 +160,12 @@ const PlatformServicesLive = Layer.unwrap( }), ); +const PullRequestMonitorLayerLive = PullRequestMonitor.layer.pipe( + Layer.provide( + PullRequestMonitor.PullRequestSnapshotFetcherLive.pipe(Layer.provide(GitHubCli.layer)), + ), +); + const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(OrchestrationReactorLive), Layer.provideMerge(ProviderRuntimeIngestionLive), @@ -165,6 +173,11 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(CheckpointReactorLive), Layer.provideMerge(ThreadDeletionReactorLive), Layer.provideMerge(AgentAwarenessRelay.layer.pipe(Layer.provide(ServerSecretStore.layer))), + Layer.provideMerge( + Layer.effectDiscard( + PullRequestMonitor.PullRequestMonitor.pipe(Effect.flatMap((monitor) => monitor.start())), + ).pipe(Layer.provide(PullRequestMonitorLayerLive)), + ), Layer.provideMerge(RuntimeReceiptBusLive), ); @@ -184,6 +197,10 @@ const ProviderLayerLive = ProviderServiceLive.pipe( ); const PersistenceLayerLive = Layer.empty.pipe(Layer.provideMerge(SqlitePersistenceLayerLive)); +// Leave the registry's OrchestrationEngineService requirement open so it is +// satisfied by the ambient runtime (and by mocks in tests) — eagerly providing +// OrchestrationLayerLive here would build a second engine + DB connection. +const MonitorRegistryLayerLive = MonitorRegistry.layer; const VcsDriverRegistryLayerLive = VcsDriverRegistry.layer.pipe( Layer.provide(VcsProjectConfig.layer), @@ -295,7 +312,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(VcsLayerLive), Layer.provideMerge(ProviderRuntimeLayerLive), Layer.provideMerge(Layer.mergeAll(TerminalLayerLive, PreviewLayerLive)), - Layer.provideMerge(PersistenceLayerLive), + Layer.provideMerge(Layer.mergeAll(PersistenceLayerLive, MonitorRegistryLayerLive)), Layer.provideMerge(Keybindings.layer), Layer.provideMerge(ProviderRegistryLive), // The instance registry is the new routing keystone — text generation, @@ -363,7 +380,10 @@ export const makeRoutesLayer = Layer.mergeAll( staticAndDevRouteLayer, websocketRpcRouteLayer, ), - McpHttpServer.layer.pipe(Layer.provide(McpSessionRegistry.layer)), + McpHttpServer.layer.pipe( + Layer.provide(McpSessionRegistry.layer), + Layer.provide(MonitorRegistryLayerLive), + ), ).pipe( Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer), diff --git a/apps/server/src/sourceControl/gitHubPullRequestMonitor.ts b/apps/server/src/sourceControl/gitHubPullRequestMonitor.ts new file mode 100644 index 00000000000..ef17444f626 --- /dev/null +++ b/apps/server/src/sourceControl/gitHubPullRequestMonitor.ts @@ -0,0 +1,440 @@ +import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import { decodeJsonResult } from "@t3tools/shared/schemaJson"; + +import type * as VcsProcess from "../vcs/VcsProcess.ts"; +import * as GitHubCli from "./GitHubCli.ts"; + +export interface PullRequestMonitorActor { + readonly login: string; + readonly type: "app" | "user"; +} + +export interface PullRequestMonitorReview { + readonly id: string; + readonly author: PullRequestMonitorActor; + readonly state: "approved" | "changes-requested" | "commented" | "dismissed" | "pending"; + readonly submittedAt: string | null; + readonly commitSha: string | null; +} + +export interface PullRequestMonitorReviewThread { + readonly id: string; + readonly author: PullRequestMonitorActor; + readonly latestCommentByViewer: boolean; + readonly body: string; + readonly path: string | null; + readonly line: number | null; + readonly createdAt: string; + readonly updatedAt: string; + readonly resolved: boolean; +} + +export interface PullRequestMonitorIssueComment { + readonly id: string; + readonly author: PullRequestMonitorActor; + readonly body: string; + readonly createdAt: string; + readonly updatedAt: string; +} + +export interface PullRequestMonitorCheckRun { + readonly id: string; + readonly name: string; + readonly status: "queued" | "in-progress" | "completed" | "unknown"; + readonly conclusion: + | "success" + | "failure" + | "neutral" + | "cancelled" + | "skipped" + | "timed-out" + | "action-required" + | "stale" + | null; + readonly startedAt: string | null; + readonly headSha: string; +} + +export interface PullRequestMonitorSnapshot { + readonly state: "open" | "closed" | "merged"; + readonly draft: boolean; + readonly headSha: string; + readonly baseRefName: string; + readonly mergeability: "mergeable" | "conflicting" | "unknown"; + readonly behindBaseBy: number | null; + readonly requiredChecksKnown: boolean; + readonly monitoringStartedAt?: string; + readonly reviews: ReadonlyArray; + readonly reviewThreads: ReadonlyArray; + readonly issueComments: ReadonlyArray; + readonly checkRuns: ReadonlyArray; +} + +const ActorSchema = Schema.Struct({ + login: Schema.String, + __typename: Schema.String, +}); +const ReviewSchema = Schema.Struct({ + id: Schema.String, + author: Schema.NullOr(ActorSchema), + state: Schema.String, + submittedAt: Schema.NullOr(Schema.String), + commit: Schema.NullOr(Schema.Struct({ oid: Schema.String })), +}); +const ThreadSchema = Schema.Struct({ + id: Schema.String, + isResolved: Schema.Boolean, + comments: Schema.Struct({ + nodes: Schema.Array( + Schema.Struct({ + author: Schema.NullOr(ActorSchema), + body: Schema.String, + path: Schema.NullOr(Schema.String), + line: Schema.NullOr(Schema.Number), + createdAt: Schema.String, + updatedAt: Schema.String, + }), + ), + }), +}); +const PullRequestPageSchema = Schema.Struct({ + data: Schema.Struct({ + viewer: Schema.Struct({ login: Schema.String }), + repository: Schema.Struct({ + pullRequest: Schema.NullOr( + Schema.Struct({ + state: Schema.String, + isDraft: Schema.Boolean, + merged: Schema.Boolean, + mergeable: Schema.String, + headRefOid: Schema.String, + baseRefName: Schema.String, + reviews: Schema.Struct({ + nodes: Schema.Array(ReviewSchema), + pageInfo: Schema.Struct({ + hasNextPage: Schema.Boolean, + endCursor: Schema.NullOr(Schema.String), + }), + }), + reviewThreads: Schema.Struct({ + nodes: Schema.Array(ThreadSchema), + pageInfo: Schema.Struct({ + hasNextPage: Schema.Boolean, + endCursor: Schema.NullOr(Schema.String), + }), + }), + }), + ), + }), + }), +}); +const RepoSchema = Schema.Struct({ nameWithOwner: Schema.String }); +const IssueCommentsSchema = Schema.Array( + Schema.Struct({ + id: Schema.Number, + user: Schema.Struct({ login: Schema.String, type: Schema.String }), + body: Schema.String, + created_at: Schema.String, + updated_at: Schema.String, + }), +); +const CheckRunsSchema = Schema.Struct({ + total_count: Schema.Number, + check_runs: Schema.Array( + Schema.Struct({ + id: Schema.Number, + name: Schema.String, + status: Schema.String, + conclusion: Schema.NullOr(Schema.String), + started_at: Schema.NullOr(Schema.String), + head_sha: Schema.String, + }), + ), +}); + +export class GitHubPullRequestMonitorDecodeError extends Schema.TaggedErrorClass()( + "GitHubPullRequestMonitorDecodeError", + { + command: Schema.Literal("gh"), + cwd: Schema.String, + detail: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export type GitHubPullRequestMonitorError = + | GitHubCli.GitHubCliError + | GitHubPullRequestMonitorDecodeError; + +const decode = >( + schema: S, + raw: string, + cwd: string, + detail: string, +) => + Effect.sync(() => decodeJsonResult(schema)(raw)).pipe( + Effect.flatMap((result) => + Result.isSuccess(result) + ? Effect.succeed(result.success) + : Effect.fail( + new GitHubPullRequestMonitorDecodeError({ + command: "gh", + cwd, + detail, + cause: result.failure, + }), + ), + ), + ); + +const actor = (value: Schema.Schema.Type | null): PullRequestMonitorActor => ({ + login: value?.login ?? "ghost", + type: value?.__typename === "Bot" ? "app" : "user", +}); + +const normalizeReviewState = (state: string): PullRequestMonitorReview["state"] => { + switch (state.toUpperCase()) { + case "APPROVED": + return "approved"; + case "CHANGES_REQUESTED": + return "changes-requested"; + case "DISMISSED": + return "dismissed"; + case "PENDING": + return "pending"; + default: + return "commented"; + } +}; + +const normalizeCheckStatus = (status: string): PullRequestMonitorCheckRun["status"] => { + if (status === "queued" || status === "completed") return status; + if (status === "in_progress") return "in-progress"; + return "unknown"; +}; + +const conclusions = new Set>([ + "success", + "failure", + "neutral", + "cancelled", + "skipped", + "timed-out", + "action-required", + "stale", +]); + +export const getPullRequestMonitorSnapshot = Effect.fn("getPullRequestMonitorSnapshot")( + function* (input: { + readonly cwd: string; + readonly pullRequestNumber: number; + }): Effect.fn.Return< + PullRequestMonitorSnapshot, + GitHubPullRequestMonitorError, + GitHubCli.GitHubCli + > { + const github = yield* GitHubCli.GitHubCli; + const repoResult = yield* github.execute({ + cwd: input.cwd, + args: ["repo", "view", "--json", "nameWithOwner"], + }); + const repo = yield* decode( + RepoSchema, + repoResult.stdout, + input.cwd, + "GitHub CLI returned invalid repository JSON.", + ); + const [owner, name] = repo.nameWithOwner.split("/"); + if (!owner || !name) { + return yield* new GitHubPullRequestMonitorDecodeError({ + command: "gh", + cwd: input.cwd, + detail: "GitHub repository nameWithOwner was invalid.", + cause: repo.nameWithOwner, + }); + } + + const reviews = new Map(); + const reviewThreads = new Map(); + let reviewCursor: string | null = null; + let threadCursor: string | null = null; + let reviewsComplete = false; + let threadsComplete = false; + let viewerLogin: string | undefined; + let metadata: + | Pick< + PullRequestMonitorSnapshot, + "state" | "draft" | "headSha" | "baseRefName" | "mergeability" + > + | undefined; + do { + const query = `query($owner:String!,$name:String!,$number:Int!,$reviews:String,$threads:String){viewer{login} repository(owner:$owner,name:$name){pullRequest(number:$number){state isDraft merged mergeable headRefOid baseRefName reviews(first:100,after:$reviews){nodes{id author{login __typename} state submittedAt commit{oid}} pageInfo{hasNextPage endCursor}} reviewThreads(first:100,after:$threads){nodes{id isResolved comments(last:1){nodes{author{login __typename} body path line createdAt updatedAt}}} pageInfo{hasNextPage endCursor}}}}}`; + const result: VcsProcess.VcsProcessOutput = yield* github.execute({ + cwd: input.cwd, + args: [ + "api", + "graphql", + "-f", + `query=${query}`, + "-F", + `owner=${owner}`, + "-F", + `name=${name}`, + "-F", + `number=${input.pullRequestNumber}`, + ...(reviewCursor ? ["-f", `reviews=${reviewCursor}`] : []), + ...(threadCursor ? ["-f", `threads=${threadCursor}`] : []), + ], + }); + const page: Schema.Schema.Type = yield* decode( + PullRequestPageSchema, + result.stdout, + input.cwd, + "GitHub GraphQL returned invalid pull request monitor JSON.", + ); + const pr: NonNullable< + Schema.Schema.Type["data"]["repository"]["pullRequest"] + > | null = page.data.repository.pullRequest; + viewerLogin ??= page.data.viewer.login; + if (!pr) { + return yield* new GitHubPullRequestMonitorDecodeError({ + command: "gh", + cwd: input.cwd, + detail: "GitHub pull request was not found.", + cause: input.pullRequestNumber, + }); + } + metadata ??= { + state: pr.merged ? "merged" : pr.state === "CLOSED" ? "closed" : "open", + draft: pr.isDraft, + headSha: pr.headRefOid, + baseRefName: pr.baseRefName, + mergeability: + pr.mergeable === "MERGEABLE" + ? "mergeable" + : pr.mergeable === "CONFLICTING" + ? "conflicting" + : "unknown", + }; + if (!reviewsComplete) { + for (const review of pr.reviews.nodes) { + reviews.set(review.id, { + id: review.id, + author: actor(review.author), + state: normalizeReviewState(review.state), + submittedAt: review.submittedAt, + commitSha: review.commit?.oid ?? null, + }); + } + reviewsComplete = !pr.reviews.pageInfo.hasNextPage; + reviewCursor = pr.reviews.pageInfo.endCursor; + } + if (!threadsComplete) { + for (const thread of pr.reviewThreads.nodes) { + const comment = thread.comments.nodes[0]; + if (comment) { + reviewThreads.set(thread.id, { + id: thread.id, + author: actor(comment.author), + latestCommentByViewer: comment.author?.login === viewerLogin, + body: comment.body, + path: comment.path, + line: comment.line, + createdAt: comment.createdAt, + updatedAt: comment.updatedAt, + resolved: thread.isResolved, + }); + } + } + threadsComplete = !pr.reviewThreads.pageInfo.hasNextPage; + threadCursor = pr.reviewThreads.pageInfo.endCursor; + } + } while (!reviewsComplete || !threadsComplete); + + const issueComments: PullRequestMonitorIssueComment[] = []; + for (let page = 1; ; page++) { + const result = yield* github.execute({ + cwd: input.cwd, + args: [ + "api", + "--method", + "GET", + `repos/${owner}/${name}/issues/${input.pullRequestNumber}/comments`, + "-f", + "per_page=100", + "-f", + `page=${page}`, + ], + }); + const entries = yield* decode( + IssueCommentsSchema, + result.stdout, + input.cwd, + "GitHub API returned invalid issue comment JSON.", + ); + issueComments.push( + ...entries + .filter((comment) => comment.user.type === "Bot") + .map((comment) => ({ + id: String(comment.id), + author: { login: comment.user.login, type: "app" as const }, + body: comment.body, + createdAt: comment.created_at, + updatedAt: comment.updated_at, + })), + ); + if (entries.length < 100) break; + } + + const checkRuns: PullRequestMonitorCheckRun[] = []; + for (let page = 1; ; page++) { + const result = yield* github.execute({ + cwd: input.cwd, + args: [ + "api", + "--method", + "GET", + `repos/${owner}/${name}/commits/${metadata.headSha}/check-runs`, + "-H", + "Accept: application/vnd.github+json", + "-f", + "per_page=100", + "-f", + `page=${page}`, + ], + }); + const response = yield* decode( + CheckRunsSchema, + result.stdout, + input.cwd, + "GitHub API returned invalid check run JSON.", + ); + checkRuns.push( + ...response.check_runs.map((check) => ({ + id: String(check.id), + name: check.name, + status: normalizeCheckStatus(check.status), + conclusion: + check.conclusion && conclusions.has(check.conclusion as never) + ? (check.conclusion as PullRequestMonitorCheckRun["conclusion"]) + : null, + startedAt: check.started_at, + headSha: check.head_sha, + })), + ); + if (response.check_runs.length < 100) break; + } + + return { + ...metadata, + behindBaseBy: null, + requiredChecksKnown: false, + reviews: [...reviews.values()], + reviewThreads: [...reviewThreads.values()], + issueComments, + checkRuns, + }; + }, +); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ab1256cddb3..83fc0c1e352 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -166,6 +166,7 @@ import { useNowMinute } from "../hooks/useNowMinute"; import { resolveAppModelSelectionForInstance } from "../modelSelection"; import { getTerminalFocusOwner } from "../lib/terminalFocus"; import { resolveNewDraftStartFromOrigin } from "../lib/chatThreadActions"; +import { useOpenPrLink } from "../lib/openPullRequestLink"; import { deriveLogicalProjectKeyFromSettings, selectProjectGroupingSettings, @@ -223,6 +224,7 @@ import { ChatHeader } from "./chat/ChatHeader"; import { PanelLayoutControls, RightPanelMaximizeControl } from "./chat/PanelLayoutControls"; import { type ExpandedImagePreview } from "./chat/ExpandedImagePreview"; import { NoActiveThreadState } from "./NoActiveThreadState"; +import { MonitorStrip } from "./MonitorStrip"; import { resolveEffectiveEnvMode, resolveLocalCheckoutBranchMismatch } from "./BranchToolbar.logic"; import { getProviderStatusBannerKey, @@ -1142,6 +1144,7 @@ function ChatViewContent(props: ChatViewProps) { const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, { reportFailure: false, }); + const endThreadMonitor = useAtomCommand(threadEnvironment.endMonitor, { reportFailure: false }); const switchGitRef = useAtomCommand(vcsEnvironment.switchRef, { reportFailure: false }); const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, { reportFailure: false, @@ -2330,6 +2333,26 @@ function ChatViewContent(props: ChatViewProps) { input: { cwd: gitStatusCwd }, }), ); + // PR monitoring ("babysit") strip: resolve the thread's PR URL for the + // ready-state "View PR"/"Merge…" affordances, and a stop handler that ends + // the monitor via the same command path the server drives. + const openPrLink = useOpenPrLink(); + const monitorPrUrl = + resolveThreadPr({ + threadBranch: activeThread?.branch ?? null, + gitStatus: gitStatusQuery.data ?? null, + hasDedicatedWorktree: activeThread?.worktreePath != null, + })?.url ?? null; + const handleStopMonitoring = useCallback(() => { + if (activeThread == null || activeThread.monitor == null) return; + void endThreadMonitor({ + environmentId: activeThread.environmentId, + input: { + threadId: activeThread.id, + blockersSummary: activeThread.monitor.blockersSummary, + }, + }); + }, [activeThread, endThreadMonitor]); const keybindings = useAtomValue(primaryServerKeybindingsAtom); const availableEditors = useAtomValue(primaryServerAvailableEditorsAtom); // Prefer an instance-id match so a custom Codex instance (e.g. @@ -5655,6 +5678,15 @@ function ChatViewContent(props: ChatViewProps) { /> + {activeThread.monitor != null ? ( + + ) : null} + setThreadError(activeThread.id, null)} diff --git a/apps/web/src/components/MonitorStrip.tsx b/apps/web/src/components/MonitorStrip.tsx new file mode 100644 index 00000000000..23e6cba66cc --- /dev/null +++ b/apps/web/src/components/MonitorStrip.tsx @@ -0,0 +1,162 @@ +import type { OrchestrationThreadMonitor } from "@t3tools/contracts"; +import { ExternalLinkIcon, GitMergeIcon } from "lucide-react"; +import { type MouseEvent, useState } from "react"; + +import { cn } from "~/lib/utils"; +import { resolveMonitorReadyLabel, resolveMonitorSidebarState } from "./Sidebar.logic"; +import { Button } from "./ui/button"; +import { + AlertDialog, + AlertDialogClose, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogPopup, + AlertDialogTitle, +} from "./ui/alert-dialog"; + +/** + * Thread-header strip for the PR monitoring ("babysit") mode. Lives at the top + * of the thread while a monitor is active (mock 4b/4c): + * + * - monitoring → cyan strip: "Monitoring PR #n" pill, the waiting-on blockers + * line, and an always-visible "Stop monitoring" button. + * - ready → emerald payoff strip: "Ready to merge" / "No known blockers" pill, + * the summary line, a "View PR" link and a confirm-to-merge "Merge…" button + * (decision D1 — never auto-merge). + * - session-ended → subtle zinc "Monitoring stopped" marker (decision D2 floor). + * + * No continuous animation and no elapsed counter (a running timer reads as + * stuck); the blockers line carries the loop's legibility instead. + */ +export function MonitorStrip({ + monitor, + prUrl, + onStopMonitoring, + onOpenPr, +}: { + monitor: OrchestrationThreadMonitor; + prUrl: string | null; + onStopMonitoring: () => void; + onOpenPr: (event: MouseEvent, prUrl: string) => void; +}) { + const [mergeConfirmOpen, setMergeConfirmOpen] = useState(false); + const state = resolveMonitorSidebarState(monitor); + // Only the meaningful, live-ish states get a strip: an actively monitoring + // PR, the ready payoff, or a dead-session floor marker. Terminal/user-stop/ + // needs-attention monitors hand off to the normal settled/status treatment. + if (state === null) return null; + + const blockers = monitor.blockersSummary.trim(); + + if (state === "stopped") { + return ( +
+
+ + + Monitoring stopped + + + PR #{monitor.prNumber} · session ended + +
+
+ ); + } + + const isReady = state === "ready"; + const handleOpenPr = (event: MouseEvent) => { + if (prUrl) onOpenPr(event, prUrl); + }; + + return ( +
+
+
+ + + {isReady + ? resolveMonitorReadyLabel(monitor.blockersSummary) + : `Monitoring PR #${monitor.prNumber}`} + + {blockers !== "" ? ( + {blockers} + ) : null} +
+ +
+ {isReady ? ( + <> + + + + ) : ( + + )} +
+
+ + + + + Merge PR #{monitor.prNumber}? + + All known blockers are clear. Merging opens the pull request on your source-control + provider to complete the merge — the decision stays with you. + + + + }>Cancel + {/* TODO: there is no client-side merge mutation today (no `gh pr + merge` path in GitActionsControl). Until the server exposes one, + confirming opens the PR so the user completes the merge there, + rather than inventing a new server mutation (out of P3 scope). */} + + + + +
+ ); +} diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 59784bf8fac..be9fdfc8ad4 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -13,6 +13,8 @@ import { isContextMenuPointerDown, isTrailingDoubleClick, orderItemsByPreferredIds, + resolveMonitorReadyLabel, + resolveMonitorSidebarState, resolveProjectStatusIndicator, resolveSidebarStageBadgeLabel, resolveThreadRowClassName, @@ -916,6 +918,121 @@ describe("resolveThreadStatusPill", () => { }), ).toMatchObject({ label: "Completed", pulse: false }); }); + + const activeMonitor = { + prNumber: 4412, + status: "monitoring" as const, + blockersSummary: "2 checks pending · waiting on Bugbot", + headSha: "a41c2f9", + wakeCount: 3, + startedAt: "2026-03-09T09:00:00.000Z", + endedAt: null, + endedReason: null, + }; + + it("shows a steady, non-pulsing Monitoring pill instead of Working during a wake turn (D6)", () => { + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, monitor: activeMonitor }, + }), + ).toMatchObject({ label: "Monitoring", pulse: false }); + }); + + it("never lets Monitoring swallow a blocked state (I4)", () => { + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, monitor: activeMonitor, hasPendingApprovals: true }, + }), + ).toMatchObject({ label: "Pending Approval" }); + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, monitor: activeMonitor, hasPendingUserInput: true }, + }), + ).toMatchObject({ label: "Awaiting Input" }); + }); + + it("suppresses the Completed pill for wake-turn completions while monitoring", () => { + expect( + resolveThreadStatusPill({ + thread: { + ...baseThread, + interactionMode: "default", + monitor: activeMonitor, + latestTurn: makeLatestTurn(), + lastVisitedAt: "2026-03-09T10:04:00.000Z", + session: { + ...baseThread.session, + status: "ready", + activeTurnId: null, + }, + }, + }), + ).toMatchObject({ label: "Monitoring" }); + }); + + it("drops the Monitoring pill once the monitor is no longer active", () => { + expect( + resolveThreadStatusPill({ + thread: { + ...baseThread, + monitor: { ...activeMonitor, status: "ready", endedReason: "ready" }, + session: { ...baseThread.session, status: "ready", activeTurnId: null }, + }, + }), + ).toBeNull(); + }); +}); + +describe("resolveMonitorSidebarState", () => { + const monitor = { + prNumber: 4412, + status: "monitoring" as const, + blockersSummary: "", + headSha: "a41c2f9", + wakeCount: 0, + startedAt: "2026-03-09T09:00:00.000Z", + endedAt: null, + endedReason: null, + }; + + it("returns null without a monitor", () => { + expect(resolveMonitorSidebarState(null)).toBeNull(); + expect(resolveMonitorSidebarState(undefined)).toBeNull(); + }); + + it("maps monitoring and ready straight through", () => { + expect(resolveMonitorSidebarState(monitor)).toBe("monitoring"); + expect(resolveMonitorSidebarState({ ...monitor, status: "ready", endedReason: "ready" })).toBe( + "ready", + ); + }); + + it("marks only a session-ended stop, deferring other terminal reasons", () => { + expect( + resolveMonitorSidebarState({ + ...monitor, + status: "stopped", + endedReason: "session-ended", + }), + ).toBe("stopped"); + expect( + resolveMonitorSidebarState({ ...monitor, status: "stopped", endedReason: "stopped" }), + ).toBeNull(); + expect( + resolveMonitorSidebarState({ + ...monitor, + status: "needs-attention", + endedReason: "needs-attention", + }), + ).toBeNull(); + }); +}); + +describe("resolveMonitorReadyLabel", () => { + it("prefers the honest fallback when required-status data is unavailable", () => { + expect(resolveMonitorReadyLabel("No known blockers")).toBe("No known blockers"); + expect(resolveMonitorReadyLabel("CI ✓ · Bugbot ✓ · Macroscope ✓")).toBe("Ready to merge"); + }); }); describe("resolveThreadRowClassName", () => { diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 7aee3100d0e..e3858f8d6b1 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -99,6 +99,7 @@ export interface ThreadStatusPill { | "Working" | "Connecting" | "Completed" + | "Monitoring" | "Pending Approval" | "Awaiting Input" | "Plan Ready"; @@ -108,8 +109,12 @@ export interface ThreadStatusPill { } const THREAD_STATUS_PRIORITY: Record = { - "Pending Approval": 5, - "Awaiting Input": 4, + "Pending Approval": 6, + "Awaiting Input": 5, + // Monitoring outranks Working/Completed so the pill stays steady through + // wake-turn fixes (decision D6), but sits below approval/input so a + // blocked state is never swallowed (invariant I4). + Monitoring: 4, Working: 3, Connecting: 3, "Plan Ready": 2, @@ -123,11 +128,41 @@ type ThreadStatusInput = Pick< | "hasPendingUserInput" | "interactionMode" | "latestTurn" + | "monitor" | "session" > & { lastVisitedAt?: string | undefined; }; +// ── PR monitoring ("babysit") sidebar/strip model ─────────────────── +// The steady sidebar-facing distillation of a thread's PR monitor. Cyan +// while monitoring, emerald at the payoff, a subtle zinc marker for a dead +// babysitter — everything else hands back to the normal status treatment. +export type MonitorSidebarState = "monitoring" | "ready" | "stopped"; + +export function resolveMonitorSidebarState( + monitor: SidebarThreadSummary["monitor"], +): MonitorSidebarState | null { + if (monitor == null) return null; + if (monitor.status === "monitoring") return "monitoring"; + if (monitor.status === "ready") return "ready"; + // D2 floor: a session-scoped monitor whose session died gets a visible + // "Monitoring stopped" marker so a dead babysitter is distinguishable from + // a quiet, live one. Other terminal reasons (terminal/stopped-by-user/ + // needs-attention) defer to the normal settled/status treatment. + if (monitor.status === "stopped" && monitor.endedReason === "session-ended") return "stopped"; + return null; +} + +// "No known blockers" is the honest fallback the server labels a ready PR +// with when required-status data isn't reliably obtainable; otherwise the +// payoff reads "Ready to merge". The distinction rides in blockersSummary. +export function resolveMonitorReadyLabel( + blockersSummary: string, +): "Ready to merge" | "No known blockers" { + return /no known blockers/i.test(blockersSummary) ? "No known blockers" : "Ready to merge"; +} + export interface ThreadJumpHintVisibilityController { sync: (shouldShow: boolean) => void; dispose: () => void; @@ -561,6 +596,19 @@ export function resolveThreadStatusPill(input: { }; } + // An active PR monitor keeps a steady "Monitoring" pill for the whole mode, + // including through wake-turn fixes — it outranks Working/Completed but, + // being placed after approval/input above, never swallows a blocked state + // (D6, invariant I4). The PR number and blockers ride in the sub-line. + if (thread.monitor != null && thread.monitor.status === "monitoring") { + return { + label: "Monitoring", + colorClass: "text-cyan-600 dark:text-cyan-300/90", + dotClass: "bg-cyan-500 dark:bg-cyan-300/90", + pulse: false, + }; + } + if (thread.session?.status === "running") { return { label: "Working", diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index e018491348e..9c467a9bd91 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -108,6 +108,8 @@ import { isTrailingDoubleClick, orderItemsByPreferredIds, resolveAdjacentThreadId, + resolveMonitorReadyLabel, + resolveMonitorSidebarState, resolveSettledTimestamp, resolveSidebarV2Status, resolveWorkingStartedAt, @@ -411,9 +413,25 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { const isSelected = useThreadSelectionStore((state) => state.selectedThreadKeys.has(threadKey)); const openPrLink = useOpenPrLink(); + // The PR monitor ("babysit") mode wraps the normal statuses: a monitoring + // row recedes like any in-flight row, a ready row demands attention like an + // unread completion, and a dead-session monitor gets a subtle floor marker. + const monitorState = resolveMonitorSidebarState(thread.monitor); + const isMonitoring = monitorState === "monitoring"; + const isMonitorReady = monitorState === "ready"; + const isMonitorStopped = monitorState === "stopped"; + // What the PR is waiting on, in place of the branch on the sub-line — the + // legibility that answers "Are you still going?" (mock 4a). No counter. + const monitorBlockers = thread.monitor?.blockersSummary?.trim() ?? ""; + const monitorSubLine = + (isMonitoring || isMonitorReady) && monitorBlockers !== "" ? monitorBlockers : null; // Same semantics as v1 (never-visited counts as read): flipping the beta - // flag must not light up every historical thread as unread. - const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); + // flag must not light up every historical thread as unread. A ready PR is + // the payoff moment — treat it as unread-prominent — while an actively + // monitoring thread suppresses the success-shaped unread from its quiet + // wake-turn completions (never a blocked state — that surfaces via status). + const isUnread = + isMonitorReady || (!isMonitoring && hasUnseenCompletion({ ...thread, lastVisitedAt })); const status = resolveSidebarV2Status(thread); // A woken thread reappears at its original position (the sort is // deliberately static), so the pill has to carry the weight. Snoozing is @@ -430,14 +448,16 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { // freshly woken. The status label keeps its hue, so waiting rows stay // findable. In-flight rows recede the same as read-ready ones (inbox-zero: // working threads aren't your problem yet) — only the colored status label - // stands out. - const isInFlight = status === "working" || status === "approval" || status === "input"; + // stands out. Monitoring rows are the ultimate "in-flight, don't look at + // me" row. + const isInFlight = + status === "working" || status === "approval" || status === "input" || isMonitoring; const shouldRecede = (status === "ready" || isInFlight) && !isUnread && !isWoke && !props.isActive && !isSelected; // Status hues follow the system-wide convention set by sidebar v1 and the // mobile Live Activity/widgets (amber approval, indigo input, sky working) // so a thread reads the same color everywhere it surfaces. - const topStatus = + const baseTopStatus = status === "working" ? { label: "Working", @@ -476,6 +496,34 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { className: "text-emerald-700 dark:text-emerald-300", } : null; + // Monitoring wraps the base status: a ready PR is the emerald payoff, an + // active monitor is one steady cyan pill through fixes (D6), a dead-session + // monitor gets the subtle zinc floor marker (D2). A genuinely blocked wake + // turn (approval/input/failed) is never swallowed — those keep the base + // pill (invariant I4). A freshly woken snoozed thread also keeps its explicit + // user-action pill instead of being swallowed by monitor state. + const isBlockedOrFailed = status === "approval" || status === "input" || status === "failed"; + const topStatus = isWoke + ? baseTopStatus + : isMonitorReady && !isBlockedOrFailed + ? { + label: resolveMonitorReadyLabel(thread.monitor?.blockersSummary ?? ""), + icon: "done" as const, + className: "text-emerald-700 dark:text-emerald-300", + } + : isMonitoring && !isBlockedOrFailed + ? { + label: "Monitoring", + icon: null, + className: "text-cyan-700 dark:text-cyan-300", + } + : isMonitorStopped && !isBlockedOrFailed + ? { + label: "Monitoring stopped", + icon: null, + className: "text-muted-foreground/80", + } + : baseTopStatus; const gitCwd = thread.worktreePath ?? props.projectCwd; const gitStatus = useEnvironmentQuery( @@ -897,7 +945,9 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { wrapper around the ticking duration would make screen readers announce every second. */} {topStatus.label} - {status === "working" ? ( + {/* No elapsed counter while monitoring — "Working 2h" + reads as stuck; the sub-line shows the wait state. */} + {status === "working" && !isMonitoring ? ( @@ -938,7 +988,16 @@ const SidebarV2Row = memo(function SidebarV2Row(props: {
{title}
- {thread.branch ? ( + {monitorSubLine !== null ? ( + + {monitorSubLine} + + ) : thread.branch ? ( {thread.branch} ) : ( diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index ad25d6544dc..246284b160c 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -2,6 +2,7 @@ import { CommandId, ORCHESTRATION_WS_METHODS, type ClientOrchestrationCommand, + type ThreadId, } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; @@ -197,6 +198,30 @@ export const unsnoozeThread: (input: UnsnoozeThreadInput) => CommandEffect = Eff }); }); +export interface EndThreadMonitorInput { + readonly threadId: ThreadId; + readonly blockersSummary: string; + readonly commandId?: CommandId; + readonly endedAt?: string; +} + +// Clients can only end a monitor as a user stop; the other end reasons +// (ready/terminal/session-ended/needs-attention) are server verdicts that the +// client-dispatchable contracts union rejects from the wire. +export const endThreadMonitor: (input: EndThreadMonitorInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.endThreadMonitor", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + type: "thread.monitor.end", + commandId: metadata.commandId, + threadId: input.threadId, + reason: "stopped", + blockersSummary: input.blockersSummary, + endedAt: metadata.createdAt, + }); +}); + export const updateThreadMetadata: (input: UpdateThreadMetadataInput) => CommandEffect = Effect.fn( "EnvironmentCommands.updateThreadMetadata", )(function* (input) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 6c128eb01ab..c60716381a5 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -6,6 +6,7 @@ import { type ArchiveThreadInput, type CreateThreadInput, type DeleteThreadInput, + type EndThreadMonitorInput, type InterruptThreadTurnInput, type RespondToThreadApprovalInput, type RespondToThreadUserInputInput, @@ -23,6 +24,7 @@ import { archiveThread, createThread, deleteThread, + endThreadMonitor, interruptThreadTurn, respondToThreadApproval, respondToThreadUserInput, @@ -44,6 +46,7 @@ export type { ArchiveThreadInput, CreateThreadInput, DeleteThreadInput, + EndThreadMonitorInput, InterruptThreadTurnInput, RespondToThreadApprovalInput, RespondToThreadUserInputInput, @@ -118,6 +121,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + endMonitor: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:end-monitor", + execute: (input: EndThreadMonitorInput) => endThreadMonitor(input), + scheduler, + concurrency, + }), updateMetadata: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:update-metadata", execute: (input: UpdateThreadMetadataInput) => updateThreadMetadata(input), diff --git a/packages/client-runtime/src/state/threadSettled.test.ts b/packages/client-runtime/src/state/threadSettled.test.ts index 421d95c6b13..e174a5c3aa7 100644 --- a/packages/client-runtime/src/state/threadSettled.test.ts +++ b/packages/client-runtime/src/state/threadSettled.test.ts @@ -24,6 +24,7 @@ function makeShell(input: { readonly activityAt: string | null; readonly sessionStatus?: "starting" | "running"; readonly pending?: "approval" | "user-input"; + readonly monitorStatus?: "monitoring" | "ready" | "stopped" | "needs-attention"; }): OrchestrationThreadShell { const threadId = ThreadId.make("thread-1"); return { @@ -67,6 +68,24 @@ function makeShell(input: { hasPendingApprovals: input.pending === "approval", hasPendingUserInput: input.pending === "user-input", hasActionableProposedPlan: false, + monitor: + input.monitorStatus === undefined + ? null + : { + prNumber: 4412, + status: input.monitorStatus, + blockersSummary: "2 checks pending · waiting on Bugbot", + headSha: "a41c2f9", + wakeCount: 1, + startedAt: "2026-04-01T00:00:00.000Z", + endedAt: input.monitorStatus === "monitoring" ? null : NOW, + endedReason: + input.monitorStatus === "monitoring" + ? null + : input.monitorStatus === "stopped" + ? "session-ended" + : input.monitorStatus, + }, }; } @@ -276,6 +295,47 @@ describe("effectiveSettled", () => { expect(effectiveSettled(boundary, { now: NOW, autoSettleAfterDays: 3 })).toBe(false); expect(effectiveSettled(stale, { now: NOW, autoSettleAfterDays: null })).toBe(false); }); + + it("holds an actively monitoring thread unsettled, outranking every auto and explicit settle signal (I2)", () => { + // A monitoring thread never auto-settles, even when stale, even on a + // merged PR, and even when the user explicitly settled it — the settle + // ends monitoring server-side (flipping status away from "monitoring") + // before this ever classifies it settled. + for (const settledOverride of [null, "settled"] as const) { + const monitoring = makeShell({ + settledOverride, + activityAt: STALE, + monitorStatus: "monitoring", + }); + expect( + effectiveSettled(monitoring, { + now: NOW, + autoSettleAfterDays: 3, + changeRequestState: "merged", + }), + ).toBe(false); + expect(effectiveSettled(monitoring, { now: NOW, autoSettleAfterDays: null })).toBe(false); + } + }); + + it("only the active 'monitoring' status blocks: ended monitors settle normally", () => { + // ready/stopped/needs-attention are terminal monitor states; the mode is + // over, so the underlying settle signals (merge, staleness, override) + // take over again. + for (const monitorStatus of ["ready", "stopped", "needs-attention"] as const) { + const ended = makeShell({ activityAt: STALE, monitorStatus }); + expect(effectiveSettled(ended, { now: NOW, autoSettleAfterDays: 3 })).toBe(true); + expect( + effectiveSettled( + makeShell({ settledOverride: "settled", activityAt: null, monitorStatus }), + { + now: NOW, + autoSettleAfterDays: null, + }, + ), + ).toBe(true); + } + }); }); describe("hasQueuedTurnStart", () => { diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index 0d077c892bb..ccceac113b8 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -231,8 +231,9 @@ export const CHANGE_REQUEST_SETTLE_IDLE_MS = 60 * 60 * 1_000; /** * Settled resolution over the server-backed settled lifecycle. Activity * blockers (pending approval/user-input, a live session, an unadjudicated - * queued turn) are checked first and hold a thread active regardless of any - * override. Past the blockers, the explicit user override (thread.settle / + * queued turn, an active PR monitor) are checked first and hold a thread + * active regardless of any override. Past the blockers, the explicit user + * override (thread.settle / * thread.unsettle commands, projected into settledOverride + settledAt) * wins in both directions; without one, a thread auto-settles on a * merged/closed PR (once idle) or inactivity past the window. The server @@ -266,6 +267,13 @@ export function effectiveSettled( Date.parse(shell.settledAt) >= Date.parse(shell.latestUserMessageAt); if (!serverAdjudicated) return false; } + // An active PR monitor holds the thread unsettled for the mode's duration + // (invariant I2): a monitoring thread lives in the active area, receded, + // until monitor.ended fires. This outranks even an explicit "settled" + // override — the user settling a monitoring thread ends monitoring + // server-side, which flips status away from "monitoring" and releases the + // blocker, so there is no client-side special-casing beyond this check. + if (shell.monitor != null && shell.monitor.status === "monitoring") return false; if (shell.settledOverride === "settled") return true; // "active" is the explicit keep-active pin: it suppresses auto-settle // until real activity clears it server-side. diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 84b7a8fa07f..bd8e0f73837 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -342,6 +342,20 @@ export const OrchestrationLatestTurn = Schema.Struct({ }); export type OrchestrationLatestTurn = typeof OrchestrationLatestTurn.Type; +export const OrchestrationThreadMonitor = Schema.Struct({ + prNumber: NonNegativeInt, + status: Schema.Literals(["monitoring", "ready", "stopped", "needs-attention"]), + blockersSummary: Schema.String, + headSha: Schema.String, + wakeCount: NonNegativeInt, + startedAt: IsoDateTime, + endedAt: Schema.NullOr(IsoDateTime), + endedReason: Schema.NullOr( + Schema.Literals(["ready", "stopped", "terminal", "session-ended", "needs-attention"]), + ), +}); +export type OrchestrationThreadMonitor = typeof OrchestrationThreadMonitor.Type; + export const OrchestrationThread = Schema.Struct({ id: ThreadId, projectId: ProjectId, @@ -367,6 +381,9 @@ export const OrchestrationThread = Schema.Struct({ // Optional so payloads from pre-snooze servers still decode. snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + monitor: Schema.optionalKey(Schema.NullOr(OrchestrationThreadMonitor)).pipe( + Schema.withDecodingDefault(Effect.succeed(null)), + ), deletedAt: Schema.NullOr(IsoDateTime), messages: Schema.Array(OrchestrationMessage), proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe( @@ -419,6 +436,9 @@ export const OrchestrationThreadShell = Schema.Struct({ settledAt: Schema.NullOr(IsoDateTime).pipe(Schema.withDecodingDefault(Effect.succeed(null))), snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + monitor: Schema.optionalKey(Schema.NullOr(OrchestrationThreadMonitor)).pipe( + Schema.withDecodingDefault(Effect.succeed(null)), + ), session: Schema.NullOr(OrchestrationSession), latestUserMessageAt: Schema.NullOr(IsoDateTime), hasPendingApprovals: Schema.Boolean, @@ -612,6 +632,44 @@ const ThreadUnsnoozeCommand = Schema.Struct({ reason: Schema.Literal("user"), }); +const ThreadMonitorStartCommand = Schema.Struct({ + type: Schema.Literal("thread.monitor.start"), + commandId: CommandId, + threadId: ThreadId, + prNumber: NonNegativeInt, + blockersSummary: Schema.String, + headSha: Schema.String, + createdAt: IsoDateTime, +}); + +const ThreadMonitorUpdateCommand = Schema.Struct({ + type: Schema.Literal("thread.monitor.update"), + commandId: CommandId, + threadId: ThreadId, + blockersSummary: Schema.String, + headSha: Schema.String, + wakeCount: NonNegativeInt, + updatedAt: IsoDateTime, +}); + +const ThreadMonitorEndCommand = Schema.Struct({ + type: Schema.Literal("thread.monitor.end"), + commandId: CommandId, + threadId: ThreadId, + reason: Schema.Literals(["ready", "stopped", "terminal", "session-ended", "needs-attention"]), + blockersSummary: Schema.String, + endedAt: IsoDateTime, +}); + +// Clients may only end a monitor as an explicit user stop. The other reasons +// (ready/terminal/session-ended/needs-attention) are server verdicts computed +// from GitHub state or lifecycle teardown — a client must not be able to +// forge them, mirroring how ThreadUnsettleCommand only carries "user". +const ClientThreadMonitorEndCommand = Schema.Struct({ + ...ThreadMonitorEndCommand.fields, + reason: Schema.Literal("stopped"), +}); + const ThreadMetaUpdateCommand = Schema.Struct({ type: Schema.Literal("thread.meta.update"), commandId: CommandId, @@ -767,6 +825,7 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, ThreadSessionStopCommand, + ClientThreadMonitorEndCommand, ]); export type DispatchableClientOrchestrationCommand = typeof DispatchableClientOrchestrationCommand.Type; @@ -792,6 +851,7 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, ThreadSessionStopCommand, + ClientThreadMonitorEndCommand, ]); export type ClientOrchestrationCommand = typeof ClientOrchestrationCommand.Type; @@ -861,6 +921,9 @@ const ThreadRevertCompleteCommand = Schema.Struct({ }); const InternalOrchestrationCommand = Schema.Union([ + ThreadMonitorStartCommand, + ThreadMonitorUpdateCommand, + ThreadMonitorEndCommand, ThreadSessionSetCommand, ThreadMessageAssistantDeltaCommand, ThreadMessageAssistantCompleteCommand, @@ -889,6 +952,9 @@ export const OrchestrationEventType = Schema.Literals([ "thread.unsettled", "thread.snoozed", "thread.unsnoozed", + "thread.monitor-started", + "thread.monitor-snapshot-updated", + "thread.monitor-ended", "thread.meta-updated", "thread.runtime-mode-set", "thread.interaction-mode-set", @@ -997,6 +1063,30 @@ export const ThreadUnsnoozedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadMonitorStartedPayload = Schema.Struct({ + threadId: ThreadId, + prNumber: NonNegativeInt, + blockersSummary: Schema.String, + headSha: Schema.String, + wakeCount: NonNegativeInt, + startedAt: IsoDateTime, +}); + +export const ThreadMonitorSnapshotUpdatedPayload = Schema.Struct({ + threadId: ThreadId, + blockersSummary: Schema.String, + headSha: Schema.String, + wakeCount: NonNegativeInt, + updatedAt: IsoDateTime, +}); + +export const ThreadMonitorEndedPayload = Schema.Struct({ + threadId: ThreadId, + reason: Schema.Literals(["ready", "stopped", "terminal", "session-ended", "needs-attention"]), + blockersSummary: Schema.String, + endedAt: IsoDateTime, +}); + export const ThreadMetaUpdatedPayload = Schema.Struct({ threadId: ThreadId, title: Schema.optional(TrimmedNonEmptyString), @@ -1184,6 +1274,21 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.unsnoozed"), payload: ThreadUnsnoozedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.monitor-started"), + payload: ThreadMonitorStartedPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.monitor-snapshot-updated"), + payload: ThreadMonitorSnapshotUpdatedPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.monitor-ended"), + payload: ThreadMonitorEndedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.meta-updated"),