From 35b744f4b8c826bb20e3d5842ac1c87a5303e3b1 Mon Sep 17 00:00:00 2001 From: roughcoder Date: Wed, 12 Aug 2026 15:34:58 +0100 Subject: [PATCH 1/5] feat(contracts): optional report amendment links on reports and envelopes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit post_report gains supersedesReportId so a session can amend an account it already gave — the case that motivated this is a queued instruction landing after the child reported, leaving a stale report as the record. Both fields are optional on SessionReport and on the thread.report.post command / thread.report-posted payload, so every already-persisted report event replays unchanged. supersededByReportId is deliberately absent from the event: at post time no superseding report exists yet, so it is derived on read paths instead. Co-Authored-By: Claude Opus 5 (1M context) --- packages/contracts/src/orchestration.test.ts | 77 +++++++++++++++++++ packages/contracts/src/orchestration.ts | 14 ++++ .../contracts/src/sessionOrchestration.ts | 28 +++++++ 3 files changed, 119 insertions(+) diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index eba1b4648b25..55a87916941d 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -511,6 +511,83 @@ it.effect("decodes thread settled and unsettled events", () => }), ); +it.effect("replays a thread.report-posted event persisted before amendments existed", () => + Effect.gen(function* () { + // Exactly the payload shape the server wrote before supersession: no + // amendment fields at all. An already-persisted event must keep decoding. + const replayed = yield* decodeOrchestrationEvent({ + sequence: 1, + eventId: "event-report-1", + aggregateKind: "thread", + aggregateId: "thread-1", + type: "thread.report-posted", + occurredAt: "2026-01-01T00:00:00.000Z", + commandId: "cmd-report-1", + causationEventId: null, + correlationId: "cmd-report-1", + metadata: {}, + payload: { + threadId: "thread-1", + report: { + reportId: "report-1", + threadId: "thread-1", + status: "success", + title: "Did the work", + summary: "All done.", + artifacts: [], + createdAt: "2026-01-01T00:00:00.000Z", + }, + updatedAt: "2026-01-01T00:00:00.000Z", + }, + }); + + if (replayed.type !== "thread.report-posted") { + assert.fail(`Expected thread.report-posted event, received ${replayed.type}.`); + } + assert.strictEqual(replayed.payload.report.supersedesReportId, undefined); + assert.strictEqual(replayed.payload.report.supersededByReportId, undefined); + // The pre-existing decoding default still applies alongside the new + // optional fields. + assert.strictEqual(replayed.payload.report.origin, "agent"); + }), +); + +it.effect("decodes an amending thread.report-posted event", () => + Effect.gen(function* () { + const amended = yield* decodeOrchestrationEvent({ + sequence: 2, + eventId: "event-report-2", + aggregateKind: "thread", + aggregateId: "thread-1", + type: "thread.report-posted", + occurredAt: "2026-01-01T01:00:00.000Z", + commandId: "cmd-report-2", + causationEventId: null, + correlationId: "cmd-report-2", + metadata: {}, + payload: { + threadId: "thread-1", + report: { + reportId: "report-2", + threadId: "thread-1", + status: "success", + title: "Also handled the late instruction", + summary: "Amended.", + artifacts: [], + supersedesReportId: "report-1", + createdAt: "2026-01-01T01:00:00.000Z", + }, + updatedAt: "2026-01-01T01:00:00.000Z", + }, + }); + + if (amended.type !== "thread.report-posted") { + assert.fail(`Expected thread.report-posted event, received ${amended.type}.`); + } + assert.strictEqual(amended.payload.report.supersedesReportId, "report-1"); + }), +); + it.effect("accepts provider-scoped model options in thread.turn.start", () => Effect.gen(function* () { const parsed = yield* decodeThreadTurnStartCommand({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 6a281b9c96a5..7f39126f9e19 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -380,6 +380,16 @@ export const SessionReport = Schema.Struct({ // Defaulted so reports stored before synthesized reports existed — all of // them agent-posted — keep decoding. origin: SessionReportOrigin.pipe(Schema.withDecodingDefault(Effect.succeed("agent" as const))), + // Amendment chain. `supersedesReportId` is written by the author (an agent + // whose account changed after it already reported, e.g. because a queued + // instruction arrived late) and travels in the persisted event; it always + // names an earlier report on the SAME thread. `supersededByReportId` is the + // other end of the same link, derived on read paths from whichever later + // report points back here — never written into an event, because at post + // time no such report exists yet. Both optional, so every report persisted + // before amendments existed keeps decoding. + supersedesReportId: Schema.optional(TrimmedNonEmptyString), + supersededByReportId: Schema.optional(TrimmedNonEmptyString), createdAt: IsoDateTime, }).check(structuredReportFieldsWithinSizeCap); export type SessionReport = typeof SessionReport.Type; @@ -1256,6 +1266,10 @@ const ThreadReportPostCommand = Schema.Struct({ // Omitted by the post_report tool; only the reactor's synthesized terminal // reports set "system". origin: Schema.optional(SessionReportOrigin), + // An earlier report on this same thread that this one amends. The toolkit + // validates the reference before dispatching; nothing here rewrites the + // superseded report — the record stays append-only. + supersedesReportId: Schema.optional(TrimmedNonEmptyString), createdAt: IsoDateTime, }).check(structuredReportFieldsWithinSizeCap); diff --git a/packages/contracts/src/sessionOrchestration.ts b/packages/contracts/src/sessionOrchestration.ts index bceec3ae6500..0639428ac257 100644 --- a/packages/contracts/src/sessionOrchestration.ts +++ b/packages/contracts/src/sessionOrchestration.ts @@ -189,6 +189,11 @@ export const SessionReportEnvelope = Schema.Struct({ artifacts: Schema.Array(SessionReportArtifact), // What the child cost, captured at report-post time. usage: Schema.optional(SessionUsageSnapshot), + // Both ends of an amendment chain, so a parent holding one envelope can + // walk to the other with read_report: the earlier report this one amends, + // and the later report that amended this one. + supersedesReportId: Schema.optional(TrimmedNonEmptyString), + supersededByReportId: Schema.optional(TrimmedNonEmptyString), createdAt: IsoDateTime, }); export type SessionReportEnvelope = typeof SessionReportEnvelope.Type; @@ -232,6 +237,12 @@ export const toSessionReportEnvelope = (report: SessionReport): SessionReportEnv validationGapsCount: report.validation?.gaps.length ?? 0, artifacts: report.artifacts, ...(report.usage !== undefined ? { usage: report.usage } : {}), + ...(report.supersedesReportId !== undefined + ? { supersedesReportId: report.supersedesReportId } + : {}), + ...(report.supersededByReportId !== undefined + ? { supersededByReportId: report.supersededByReportId } + : {}), createdAt: report.createdAt, }; if (report.summary.length <= SESSION_REPORT_INLINE_MAX_CHARS) { @@ -334,10 +345,22 @@ export const ReadReportResult = Schema.Struct({ ), artifacts: Schema.Array(SessionReportArtifact), usage: Schema.optional(SessionUsageSnapshot), + // Amendment chain, both directions. + supersedesReportId: Schema.optional(TrimmedNonEmptyString), + supersededByReportId: Schema.optional(TrimmedNonEmptyString), + // Present only when this report has been superseded. A caller paging an old + // body must not have to notice an id field to learn the record moved on, so + // the fact is also stated in prose it cannot miss. + supersededNotice: Schema.optional(Schema.String), createdAt: IsoDateTime, }); export type ReadReportResult = typeof ReadReportResult.Type; +// The prose half of the marker above. One builder so read_report has a single +// wording, and tests assert against the same string the caller sees. +export const supersededReportNotice = (supersededByReportId: string): string => + `SUPERSEDED: this report was amended by a newer report on the same session. Read reportId "${supersededByReportId}" for the current account of the work; treat anything below as out of date.`; + export const PingSessionInput = Schema.Struct({ threadId: ThreadId, }); @@ -434,6 +457,11 @@ export const PostReportInput = Schema.Struct({ completionPercent: Schema.optional( Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).check(Schema.isLessThanOrEqualTo(100)), ), + // reportId of an earlier report by THIS session that this one amends and + // replaces as the session's current account. Must name a report on this + // same thread. The superseded report is kept and stays readable, flagged as + // superseded; this one becomes the latest. + supersedesReportId: Schema.optional(TrimmedNonEmptyString), }).check(structuredReportFieldsWithinSizeCap); export type PostReportInput = typeof PostReportInput.Type; From 84a33272c6c9a93161fe895592780f1f704b9d82 Mon Sep 17 00:00:00 2001 From: roughcoder Date: Wed, 12 Aug 2026 15:35:15 +0100 Subject: [PATCH 2/5] feat(server): persist report amendments and derive the reverse link Migration 048 adds projection_thread_reports.supersedes_report_id plus the index every read path uses to resolve the other direction. Only the forward link is stored: an amendment never rewrites the row it supersedes, so the projection stays append-only and "who superseded me" is a correlated subquery rather than a flag that could drift out of sync. Migration id 48 leaves a gap at 47, which a sibling wave-2 slice reserves; ids only have to be ordered, not contiguous. Co-Authored-By: Claude Opus 5 (1M context) --- .../Layers/ProjectionPipeline.ts | 1 + .../Layers/ProjectionSnapshotQuery.ts | 25 ++++++ .../src/orchestration/decider.reports.test.ts | 29 ++++++ apps/server/src/orchestration/decider.ts | 6 ++ .../Layers/ProjectionRepositories.test.ts | 89 +++++++++++++++++++ .../Layers/ProjectionThreadReports.ts | 34 +++++++ apps/server/src/persistence/Migrations.ts | 2 + ...8_ProjectionThreadReportSupersedes.test.ts | 78 ++++++++++++++++ .../048_ProjectionThreadReportSupersedes.ts | 22 +++++ .../Services/ProjectionThreadReports.ts | 7 ++ 10 files changed, 293 insertions(+) create mode 100644 apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.test.ts create mode 100644 apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.ts diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 61bd15dfe4c3..bf82da4e508d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1109,6 +1109,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti ? { usage: event.payload.report.usage } : {}), origin: event.payload.report.origin, + supersedesReportId: event.payload.report.supersedesReportId ?? null, createdAt: event.payload.report.createdAt, }); return; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 2c4d4d53561d..82b4bca83659 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -13,6 +13,7 @@ import { OrchestrationThreadDetailSnapshot, ProjectScript, SessionReportArtifact, + TrimmedNonEmptyString, TurnId, type OrchestrationCheckpointSummary, type OrchestrationLatestTurn, @@ -96,10 +97,14 @@ const ProjectionThreadReportDbRowSchema = ProjectionThreadReport.mapFields((fiel "summary", "abstract", "origin", + "supersedesReportId", "createdAt", ]), { artifacts: Schema.fromJsonString(Schema.Array(SessionReportArtifact)), + // Derived reverse amendment link; NULL unless a later report on the + // thread names this one. + supersededByReportId: Schema.NullOr(TrimmedNonEmptyString), // Optional findings/validation/recommendation/completionPercent, // stored together as one JSON column. Decoded leniently (see // decodeStructuredReportFields) rather than through the schema, so a @@ -392,6 +397,10 @@ function mapReportRow( artifacts: row.artifacts, ...decodeStructuredReportFields(row.structuredJson), origin: row.origin, + ...(row.supersedesReportId !== null ? { supersedesReportId: row.supersedesReportId } : {}), + ...(row.supersededByReportId !== null + ? { supersededByReportId: row.supersededByReportId } + : {}), createdAt: row.createdAt, }; } @@ -631,6 +640,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { artifacts_json AS "artifacts", structured_json AS "structuredJson", origin, + supersedes_report_id AS "supersedesReportId", + ( + SELECT amendment.report_id + FROM projection_thread_reports AS amendment + WHERE amendment.supersedes_report_id = projection_thread_reports.report_id + ORDER BY amendment.created_at ASC, amendment.report_id ASC + LIMIT 1 + ) AS "supersededByReportId", created_at AS "createdAt" FROM projection_thread_reports ORDER BY thread_id ASC, created_at ASC, report_id ASC @@ -1136,6 +1153,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { artifacts_json AS "artifacts", structured_json AS "structuredJson", origin, + supersedes_report_id AS "supersedesReportId", + ( + SELECT amendment.report_id + FROM projection_thread_reports AS amendment + WHERE amendment.supersedes_report_id = projection_thread_reports.report_id + ORDER BY amendment.created_at ASC, amendment.report_id ASC + LIMIT 1 + ) AS "supersededByReportId", created_at AS "createdAt" FROM projection_thread_reports WHERE thread_id = ${threadId} diff --git a/apps/server/src/orchestration/decider.reports.test.ts b/apps/server/src/orchestration/decider.reports.test.ts index d7c38b490fd9..44693274debb 100644 --- a/apps/server/src/orchestration/decider.reports.test.ts +++ b/apps/server/src/orchestration/decider.reports.test.ts @@ -167,6 +167,35 @@ it.layer(NodeServices.layer)("report decider", (it) => { }), ); + it.effect("carries an amendment link through to the event payload", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.report.post", + commandId: CommandId.make("cmd-report-post-amendment"), + threadId: ThreadId.make("thread-1"), + reportId: "report-amendment", + status: "success", + title: "Also handled the late instruction", + summary: "The queued instruction arrived after the first report; it is done now.", + artifacts: [], + supersedesReportId: "report-1", + createdAt: POSTED_AT, + }, + readModel, + }); + const event = Array.isArray(result) ? result[0] : result; + + expect(event.type).toBe("thread.report-posted"); + if (event.type === "thread.report-posted") { + expect(event.payload.report.supersedesReportId).toBe("report-1"); + // Only the forward link is recorded: the superseded report keeps its + // own event, and the reverse link is derived when reports are read. + expect(event.payload.report.supersededByReportId).toBeUndefined(); + } + }), + ); + it.effect("rejects a report for an unknown thread", () => Effect.gen(function* () { const result = yield* decideOrchestrationCommand({ diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 75e167755f3e..a5e529fed352 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1496,6 +1496,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" // post_report never sets an origin: an agent-posted report is the // default, and only the reactor claims "system". origin: command.origin ?? "agent", + // Amendments append: the superseded report keeps its own event, + // and only this forward link is recorded. The reverse link is + // derived when reports are read back. + ...(command.supersedesReportId !== undefined + ? { supersedesReportId: command.supersedesReportId } + : {}), createdAt: command.createdAt, }, updatedAt: command.createdAt, diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index 1edb1ed9d6ef..fa323b54deac 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -7,18 +7,39 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "./Sqlite.ts"; import { ProjectionProjectRepositoryLive } from "./ProjectionProjects.ts"; +import { ProjectionThreadReportRepositoryLive } from "./ProjectionThreadReports.ts"; import { ProjectionThreadRepositoryLive } from "./ProjectionThreads.ts"; import { ProjectionProjectRepository } from "../Services/ProjectionProjects.ts"; +import { + type ProjectionThreadReport, + ProjectionThreadReportRepository, +} from "../Services/ProjectionThreadReports.ts"; import { ProjectionThreadRepository } from "../Services/ProjectionThreads.ts"; const projectionRepositoriesLayer = it.layer( Layer.mergeAll( ProjectionProjectRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)), ProjectionThreadRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)), + ProjectionThreadReportRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)), SqlitePersistenceMemory, ), ); +const report = ( + overrides: Partial & Pick, +): ProjectionThreadReport => ({ + threadId: ThreadId.make("thread-reports"), + status: "success", + title: "Did the work", + summary: "All done.", + abstract: null, + artifacts: [], + origin: "agent", + supersedesReportId: null, + createdAt: "2026-08-12T00:00:00.000Z", + ...overrides, +}); + projectionRepositoriesLayer("Projection repositories", (it) => { it.effect("stores SQL NULL for missing project model options", () => Effect.gen(function* () { @@ -203,4 +224,72 @@ projectionRepositoriesLayer("Projection repositories", (it) => { assert.strictEqual(updated?.pinnedAt, null); }), ); + + it.effect("resolves an amendment chain from both ends", () => + Effect.gen(function* () { + const reports = yield* ProjectionThreadReportRepository; + + yield* reports.upsert( + report({ + reportId: "report-original", + summary: "Shipped the feature.", + createdAt: "2026-08-12T01:00:00.000Z", + }), + ); + yield* reports.upsert( + report({ + reportId: "report-amendment", + summary: "Shipped the feature, plus the late instruction.", + supersedesReportId: "report-original", + createdAt: "2026-08-12T02:00:00.000Z", + }), + ); + + // Read from the new end: it names what it replaced, and nothing has + // replaced it. + const amendment = Option.getOrNull( + yield* reports.findByReportId({ reportId: "report-amendment" }), + ); + assert.strictEqual(amendment?.supersedesReportId, "report-original"); + assert.strictEqual(amendment?.supersededByReportId, undefined); + + // Read from the old end: the reverse link is derived, not stored, so it + // appears without the original row ever being rewritten. + const original = Option.getOrNull( + yield* reports.findByReportId({ reportId: "report-original" }), + ); + assert.strictEqual(original?.supersedesReportId, null); + assert.strictEqual(original?.supersededByReportId, "report-amendment"); + // Append-only: the superseded report keeps its own body. + assert.strictEqual(original?.summary, "Shipped the feature."); + + // The amendment is the thread's latest report, and the list carries the + // same links as the by-id reads. + const listed = yield* reports.listByThreadId({ + threadId: ThreadId.make("thread-reports"), + }); + assert.deepStrictEqual( + listed.map((entry) => entry.reportId), + ["report-original", "report-amendment"], + ); + assert.strictEqual(listed.at(-1)?.reportId, "report-amendment"); + assert.strictEqual(listed[0]?.supersededByReportId, "report-amendment"); + }), + ); + + it.effect("leaves an unamended report with no supersession links", () => + Effect.gen(function* () { + const reports = yield* ProjectionThreadReportRepository; + + yield* reports.upsert( + report({ reportId: "report-standalone", threadId: ThreadId.make("thread-standalone") }), + ); + + const persisted = Option.getOrNull( + yield* reports.findByReportId({ reportId: "report-standalone" }), + ); + assert.strictEqual(persisted?.supersedesReportId, null); + assert.strictEqual(persisted?.supersededByReportId, undefined); + }), + ); }); diff --git a/apps/server/src/persistence/Layers/ProjectionThreadReports.ts b/apps/server/src/persistence/Layers/ProjectionThreadReports.ts index 68895fc928f4..e4555bb66dfa 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadReports.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadReports.ts @@ -39,9 +39,20 @@ const ProjectionThreadReportDbRowSchema = Schema.Struct({ // malformed blob can never fail the whole row. structuredJson: Schema.NullOr(Schema.String), origin: SessionReportOrigin, + supersedesReportId: Schema.NullOr(TrimmedNonEmptyString), + // Resolved by the reverse-link subquery every read path shares; NULL when + // no later report amends this one. + supersededByReportId: Schema.NullOr(TrimmedNonEmptyString), createdAt: IsoDateTime, }); +// Every read path selects the reverse amendment link with the same +// correlated subquery (repeated inline because a shared string would become a +// bound parameter, not SQL): the earliest report naming this one as its +// predecessor. Two rows could in principle name the same predecessor (a +// retried amendment), and taking the earliest keeps the chain a caller walks +// stable instead of scan-order dependent. + // Single row→report mapping shared by every read path (list and by-id), so a // column added here can never be silently dropped by one of them. function mapReportRow( @@ -56,6 +67,10 @@ function mapReportRow( abstract: row.abstract, artifacts: row.artifacts, origin: row.origin, + supersedesReportId: row.supersedesReportId, + ...(row.supersededByReportId !== null + ? { supersededByReportId: row.supersededByReportId } + : {}), ...decodeStructuredReportFields(row.structuredJson), createdAt: row.createdAt, }; @@ -110,6 +125,7 @@ const makeProjectionThreadReportRepository = Effect.gen(function* () { artifacts_json, structured_json, origin, + supersedes_report_id, created_at ) VALUES ( @@ -122,6 +138,7 @@ const makeProjectionThreadReportRepository = Effect.gen(function* () { ${JSON.stringify(row.artifacts)}, ${encodeStructured(row)}, ${row.origin}, + ${row.supersedesReportId}, ${row.createdAt} ) ON CONFLICT (report_id) @@ -134,6 +151,7 @@ const makeProjectionThreadReportRepository = Effect.gen(function* () { artifacts_json = excluded.artifacts_json, structured_json = excluded.structured_json, origin = excluded.origin, + supersedes_report_id = excluded.supersedes_report_id, created_at = excluded.created_at `, }); @@ -152,6 +170,14 @@ const makeProjectionThreadReportRepository = Effect.gen(function* () { artifacts_json AS "artifacts", structured_json AS "structuredJson", origin, + supersedes_report_id AS "supersedesReportId", + ( + SELECT amendment.report_id + FROM projection_thread_reports AS amendment + WHERE amendment.supersedes_report_id = projection_thread_reports.report_id + ORDER BY amendment.created_at ASC, amendment.report_id ASC + LIMIT 1 + ) AS "supersededByReportId", created_at AS "createdAt" FROM projection_thread_reports WHERE thread_id = ${threadId} @@ -173,6 +199,14 @@ const makeProjectionThreadReportRepository = Effect.gen(function* () { artifacts_json AS "artifacts", structured_json AS "structuredJson", origin, + supersedes_report_id AS "supersedesReportId", + ( + SELECT amendment.report_id + FROM projection_thread_reports AS amendment + WHERE amendment.supersedes_report_id = projection_thread_reports.report_id + ORDER BY amendment.created_at ASC, amendment.report_id ASC + LIMIT 1 + ) AS "supersededByReportId", created_at AS "createdAt" FROM projection_thread_reports WHERE report_id = ${reportId} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 9682176520ad..8ba1d48a00b8 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -60,6 +60,7 @@ import Migration0044 from "./Migrations/044_ProjectionThreadSessionStopAudit.ts" import Migration0045 from "./Migrations/045_ProjectionThreadReportsOrigin.ts"; import Migration0046 from "./Migrations/046_ProjectionThreadReportAbstract.ts"; import Migration0047 from "./Migrations/047_ProjectionQueuedTurnReceipts.ts"; +import Migration0048 from "./Migrations/048_ProjectionThreadReportSupersedes.ts"; /** * Migration loader with all migrations defined inline. @@ -119,6 +120,7 @@ export const migrationEntries = [ [45, "ProjectionThreadReportsOrigin", Migration0045], [46, "ProjectionThreadReportAbstract", Migration0046], [47, "ProjectionQueuedTurnReceipts", Migration0047], + [48, "ProjectionThreadReportSupersedes", Migration0048], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.test.ts b/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.test.ts new file mode 100644 index 000000000000..539d3cb19d56 --- /dev/null +++ b/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.test.ts @@ -0,0 +1,78 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("048_ProjectionThreadReportSupersedes", (it) => { + it.effect("adds a nullable supersedes link and its lookup index", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 48 }); + + const columns = yield* sql<{ + readonly name: string; + readonly notnull: number; + readonly dflt_value: string | null; + }>` + PRAGMA table_info(projection_thread_reports) + `; + const column = columns.find((entry) => entry.name === "supersedes_report_id"); + assert.isDefined(column); + // Nullable with no default: reports that predate amendments — every one + // already stored — supersede nothing, and must keep decoding. + assert.strictEqual(column?.notnull, 0); + assert.isNull(column?.dflt_value ?? null); + + const indexes = yield* sql<{ readonly name: string }>` + PRAGMA index_list(projection_thread_reports) + `; + assert.isTrue( + indexes.some( + (entry) => entry.name === "idx_projection_thread_reports_supersedes_report_id", + ), + ); + }), + ); + + it.effect("leaves reports written before the migration superseding nothing", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 46 }); + yield* sql` + INSERT INTO projection_thread_reports ( + report_id, + thread_id, + status, + title, + summary, + artifacts_json, + created_at + ) + VALUES ( + 'report-pre-migration', + 'thread-1', + 'success', + 'Did the work', + 'All done.', + '[]', + '2026-08-12T00:00:00.000Z' + ) + `; + + yield* runMigrations({ toMigrationInclusive: 48 }); + + const rows = yield* sql<{ readonly supersedes_report_id: string | null }>` + SELECT supersedes_report_id + FROM projection_thread_reports + WHERE report_id = 'report-pre-migration' + `; + assert.strictEqual(rows.length, 1); + assert.isNull(rows[0]?.supersedes_report_id ?? null); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.ts b/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.ts new file mode 100644 index 000000000000..82255999ae92 --- /dev/null +++ b/apps/server/src/persistence/Migrations/048_ProjectionThreadReportSupersedes.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Forward link only: a report names the earlier report it amends. The + // reverse link ("who superseded me") is derived on read from this column, + // so an amendment never rewrites the row it supersedes and the projection + // stays append-only. + yield* sql` + ALTER TABLE projection_thread_reports + ADD COLUMN supersedes_report_id TEXT + `; + + // Every read path resolves the reverse link with a lookup by this column, + // including full-table scans of the reports projection. + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_thread_reports_supersedes_report_id + ON projection_thread_reports(supersedes_report_id) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadReports.ts b/apps/server/src/persistence/Services/ProjectionThreadReports.ts index 39cb83c8885b..1a591c2e0ba4 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadReports.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadReports.ts @@ -34,6 +34,13 @@ export const ProjectionThreadReport = Schema.Struct({ // as the contract-level SessionReport/PostReportInput. ...SessionReportStructured.fields, origin: SessionReportOrigin, + // Stored forward link: the earlier report on this thread that this row + // amends. + supersedesReportId: Schema.NullOr(TrimmedNonEmptyString), + // Reverse link, derived on read (the row that points back at this one) and + // never written — hence optional, so writers cannot be asked to supply a + // value only the read path knows. + supersededByReportId: Schema.optional(TrimmedNonEmptyString), createdAt: IsoDateTime, }); export type ProjectionThreadReport = typeof ProjectionThreadReport.Type; From 4fb8493dacf52c160d01dd9ec0bde8b113242e78 Mon Sep 17 00:00:00 2001 From: roughcoder Date: Wed, 12 Aug 2026 15:35:28 +0100 Subject: [PATCH 3/5] feat(sessions): amend reports via post_report supersedesReportId MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit post_report validates the reference before dispatching: it must name a report on the calling thread. A report is a session's account of its own work, so amending another session's report is refused — with the same message as an unknown id, so the denial cannot double as a probe for report ids elsewhere. A dangling link would be worse than a rejection: no reader could follow it. Both ends travel outward. An amending report's parent notification leads with "AMENDED report (supersedes ...)" before the summary, because a parent that already acted on the superseded report has to see that first. read_report on a superseded report still serves its body, plus supersededByReportId and a supersededNotice sentence — a caller paging an old report cannot be relied on to notice a field it was not looking for. The spawned-session instructions now state the rule directly: an instruction arriving after you reported means post an amending report, never claim retroactive compliance. Co-Authored-By: Claude Opus 5 (1M context) --- .../mcp/toolkits/sessions/handlers.test.ts | 296 +++++++++++++++++- .../src/mcp/toolkits/sessions/handlers.ts | 44 ++- .../server/src/mcp/toolkits/sessions/tools.ts | 4 +- .../Layers/SessionSpawnReactor.test.ts | 28 ++ .../Layers/SessionSpawnReactor.ts | 11 +- 5 files changed, 372 insertions(+), 11 deletions(-) diff --git a/apps/server/src/mcp/toolkits/sessions/handlers.test.ts b/apps/server/src/mcp/toolkits/sessions/handlers.test.ts index 227c1d361174..360b12031fc0 100644 --- a/apps/server/src/mcp/toolkits/sessions/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/sessions/handlers.test.ts @@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, GitCommandError, + type OrchestrationCommand, type OrchestrationThreadShell, type ProjectId, ProviderInstanceId, @@ -9,8 +10,10 @@ import { ReadReportInput, SESSION_REPORT_INLINE_MAX_CHARS, SessionOrchestrationDeniedError, + SessionOrchestrationInvalidInputError, type SessionReport, type SessionUsageSnapshot, + supersededReportNotice, ThreadId, toSessionReportEnvelope, } from "@t3tools/contracts"; @@ -26,7 +29,11 @@ import * as OrchestrationEngine from "../../../orchestration/Services/Orchestrat import * as ProjectionSnapshotQuery from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; import * as ThreadTurnBootstrap from "../../../orchestration/ThreadTurnBootstrap.ts"; import { PersistenceSqlError } from "../../../persistence/Errors.ts"; -import { ProjectionThreadReportRepository } from "../../../persistence/Services/ProjectionThreadReports.ts"; +import { + type ProjectionThreadReport, + ProjectionThreadReportRepository, + type ProjectionThreadReportRepositoryShape, +} from "../../../persistence/Services/ProjectionThreadReports.ts"; import { ProviderSessionDirectoryPersistenceError } from "../../../provider/Errors.ts"; import * as ProviderRegistry from "../../../provider/Services/ProviderRegistry.ts"; import { ProviderSessionDirectory } from "../../../provider/Services/ProviderSessionDirectory.ts"; @@ -45,6 +52,7 @@ import { resolveSendToSessionDelivery, resolveSessionCheckout, sliceReportBody, + SUPERSEDES_REPORT_NOT_FOUND_MESSAGE, validateSpawnCheckoutInput, } from "./handlers.ts"; @@ -391,6 +399,20 @@ describe("toSessionReportEnvelope", () => { expect(envelope.summaryChars).toBe(SESSION_REPORT_INLINE_MAX_CHARS + 1); }); + it("carries both ends of an amendment chain so a parent can follow it", () => { + const superseded = toSessionReportEnvelope( + makeReport({ reportId: "report-original", supersededByReportId: "report-amendment" }), + ); + expect(superseded.supersededByReportId).toBe("report-amendment"); + expect(superseded.supersedesReportId).toBeUndefined(); + + const amendment = toSessionReportEnvelope( + makeReport({ reportId: "report-amendment", supersedesReportId: "report-original" }), + ); + expect(amendment.supersedesReportId).toBe("report-original"); + expect(amendment.supersededByReportId).toBeUndefined(); + }); + it("falls back to a truncated summary head when no abstract was posted", () => { const report = makeReport({ summary: "z".repeat(SESSION_REPORT_INLINE_MAX_CHARS + 1) }); const envelope = toSessionReportEnvelope(report); @@ -631,6 +653,14 @@ const childShell = { spawnedByThreadId: parentThreadId, } satisfies OrchestrationThreadShell; +// The calling session itself. post_report and read_report both resolve the +// caller's own shell before doing anything, so the default lookup has to know +// about it as well as the child. +const parentShell = { + ...baseShell, + id: parentThreadId, +} satisfies OrchestrationThreadShell; + const invocationScope = { environmentId: EnvironmentId.make("environment-1"), threadId: parentThreadId, @@ -654,6 +684,13 @@ const runHandler = ( getLatestUsageActivity?: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape["getLatestUsageActivity"]; getThreadTurnCount?: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape["getThreadTurnCount"]; listBindings?: ProviderSessionDirectory["Service"]["listBindings"]; + findByReportId?: ProjectionThreadReportRepositoryShape["findByReportId"]; + listByThreadId?: ProjectionThreadReportRepositoryShape["listByThreadId"]; + // Supplied together by the write-path tests (post_report): the default + // pair dies on use, which is what proves the read-only tools never + // dispatch. + dispatch?: OrchestrationEngine.OrchestrationEngineShape["dispatch"]; + enqueueCommand?: ServerRuntimeStartup.ServerRuntimeStartup["Service"]["enqueueCommand"]; } = {}, ) => Effect.gen(function* () { @@ -667,7 +704,9 @@ const runHandler = ( serverSettingsLayerTest({ enableSessionOrchestration: true }), Layer.mock(OrchestrationEngine.OrchestrationEngineService)({ readEvents: () => Stream.empty, - dispatch: () => Effect.die("engine.dispatch must not be called by a read-only tool"), + dispatch: + overrides.dispatch ?? + (() => Effect.die("engine.dispatch must not be called by a read-only tool")), streamDomainEvents: Stream.empty, latestSequence: Effect.succeed(0), }), @@ -676,7 +715,13 @@ const runHandler = ( getThreadShellById: overrides.getThreadShellById ?? ((id) => - Effect.succeed(id === childThreadId ? Option.some(childShell) : Option.none())), + Effect.succeed( + id === childThreadId + ? Option.some(childShell) + : id === parentThreadId + ? Option.some(parentShell) + : Option.none(), + )), getThreadHasReport: overrides.getThreadHasReport ?? (() => Effect.succeed(false)), getLastAssistantMessage: overrides.getLastAssistantMessage ?? (() => Effect.succeed(Option.none())), @@ -694,10 +739,14 @@ const runHandler = ( Layer.mock(GitWorkflowService.GitWorkflowService)({}), // Not exercised by ping_session/read_session (only read_report/post_report // touch it); unused methods die if called. - Layer.mock(ProjectionThreadReportRepository)({}), + Layer.mock(ProjectionThreadReportRepository)({ + ...(overrides.findByReportId ? { findByReportId: overrides.findByReportId } : {}), + ...(overrides.listByThreadId ? { listByThreadId: overrides.listByThreadId } : {}), + }), Layer.mock(ServerRuntimeStartup.ServerRuntimeStartup)({ - enqueueCommand: () => - Effect.die("startup.enqueueCommand must not be called by a read-only tool"), + enqueueCommand: + overrides.enqueueCommand ?? + (() => Effect.die("startup.enqueueCommand must not be called by a read-only tool")), }), ), ), @@ -840,6 +889,241 @@ describe("ping_session (handler)", () => { ); }); +const projectedReport = ( + overrides: Partial & Pick, +): ProjectionThreadReport => ({ + // post_report and read_report both run as the calling thread, which in this + // harness is the parent. + threadId: parentThreadId, + status: "success", + title: "Did the work", + summary: "All done.", + abstract: null, + artifacts: [], + origin: "agent", + supersedesReportId: null, + createdAt: now, + ...overrides, +}); + +describe("post_report supersession (handler)", () => { + // Captures what post_report dispatched, so the tests can assert the + // amendment link reached the command rather than only the tool result. + const capturingDispatch = (captured: Array) => ({ + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + captured.push(command); + return { sequence: 1 }; + }), + enqueueCommand: (effect: Effect.Effect) => effect, + }); + + it.effect("carries a valid supersedesReportId into the command and the result", () => + Effect.gen(function* () { + const captured: Array = []; + const result = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Amended: also did the late instruction", + summary: "The queued instruction arrived after the first report; it is done now.", + supersedesReportId: "report-original", + }), + { + ...capturingDispatch(captured), + findByReportId: () => + Effect.succeed(Option.some(projectedReport({ reportId: "report-original" }))), + }, + ); + + expect(result.supersedesReportId).toBe("report-original"); + expect(captured).toHaveLength(1); + expect(captured[0]).toMatchObject({ + type: "thread.report.post", + supersedesReportId: "report-original", + }); + }), + ); + + it.effect("refuses a supersedesReportId that names no report, without dispatching", () => + Effect.gen(function* () { + const captured: Array = []; + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Amended", + summary: "Amending a report that does not exist.", + supersedesReportId: "report-nonexistent", + }), + { + ...capturingDispatch(captured), + findByReportId: () => Effect.succeed(Option.none()), + }, + ).pipe(Effect.flip); + + expect(error).toBeInstanceOf(SessionOrchestrationInvalidInputError); + expect(error.message).toBe(SUPERSEDES_REPORT_NOT_FOUND_MESSAGE); + // A dangling amendment link must never reach the event log. + expect(captured).toHaveLength(0); + }), + ); + + it.effect("refuses a supersedesReportId belonging to another thread", () => + Effect.gen(function* () { + const captured: Array = []; + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Amended", + summary: "Amending someone else's report.", + supersedesReportId: "report-of-another-thread", + }), + { + ...capturingDispatch(captured), + findByReportId: () => + Effect.succeed( + Option.some( + projectedReport({ + reportId: "report-of-another-thread", + threadId: childThreadId, + }), + ), + ), + }, + ).pipe(Effect.flip); + + expect(error).toBeInstanceOf(SessionOrchestrationInvalidInputError); + // Same message as the unknown-id case: a report is a session's account + // of its own work, and the denial must not double as a probe for which + // report ids exist on other threads. + expect(error.message).toBe(SUPERSEDES_REPORT_NOT_FOUND_MESSAGE); + expect(captured).toHaveLength(0); + }), + ); + + it.effect("posts an ordinary report without touching the report repository", () => + Effect.gen(function* () { + const captured: Array = []; + // findByReportId is left unmocked here, so it dies if called: a report + // with no supersedesReportId must not pay for a lookup. + const result = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Did the work", + summary: "All done.", + }), + capturingDispatch(captured), + ); + + expect(result.supersedesReportId).toBeUndefined(); + expect(captured[0]).toMatchObject({ type: "thread.report.post" }); + expect(captured[0]).not.toHaveProperty("supersedesReportId"); + }), + ); +}); + +describe("read_report supersession (handler)", () => { + it.effect("marks a superseded report and points at the report that replaced it", () => + Effect.gen(function* () { + const result = yield* runHandler( + (handlers) => handlers.read_report({ reportId: "report-original" }), + { + findByReportId: () => + Effect.succeed( + Option.some( + projectedReport({ + reportId: "report-original", + summary: "Shipped the feature.", + supersededByReportId: "report-amendment", + }), + ), + ), + }, + ); + + expect(result.supersededByReportId).toBe("report-amendment"); + // Prose as well as an id: a caller paging an old body cannot be relied + // on to notice a field it was not looking for. + expect(result.supersededNotice).toBe(supersededReportNotice("report-amendment")); + expect(result.supersededNotice).toContain("report-amendment"); + // Append-only: the superseded body is still served. + expect(result.body).toBe("Shipped the feature."); + }), + ); + + it.effect("reads the chain from the new end without a superseded marker", () => + Effect.gen(function* () { + const result = yield* runHandler( + (handlers) => handlers.read_report({ reportId: "report-amendment" }), + { + findByReportId: () => + Effect.succeed( + Option.some( + projectedReport({ + reportId: "report-amendment", + summary: "Shipped the feature, plus the late instruction.", + supersedesReportId: "report-original", + }), + ), + ), + }, + ); + + expect(result.supersedesReportId).toBe("report-original"); + expect(result.supersededByReportId).toBeUndefined(); + expect(result.supersededNotice).toBeUndefined(); + }), + ); + + it.effect("returns the amendment, unmarked, as the thread's latest report", () => + Effect.gen(function* () { + const result = yield* runHandler( + (handlers) => handlers.read_report({ threadId: parentThreadId }), + { + listByThreadId: () => + Effect.succeed([ + projectedReport({ + reportId: "report-original", + summary: "Shipped the feature.", + supersededByReportId: "report-amendment", + createdAt: "2026-08-12T01:00:00.000Z", + }), + projectedReport({ + reportId: "report-amendment", + summary: "Shipped the feature, plus the late instruction.", + supersedesReportId: "report-original", + createdAt: "2026-08-12T02:00:00.000Z", + }), + ]), + }, + ); + + expect(result.reportId).toBe("report-amendment"); + expect(result.supersedesReportId).toBe("report-original"); + expect(result.supersededNotice).toBeUndefined(); + }), + ); + + it.effect("leaves an unamended report unmarked", () => + Effect.gen(function* () { + const result = yield* runHandler( + (handlers) => handlers.read_report({ reportId: "report-standalone" }), + { + findByReportId: () => + Effect.succeed(Option.some(projectedReport({ reportId: "report-standalone" }))), + }, + ); + + expect(result.supersedesReportId).toBeUndefined(); + expect(result.supersededByReportId).toBeUndefined(); + expect(result.supersededNotice).toBeUndefined(); + }), + ); +}); + describe("read_session (handler)", () => { it.effect("degrades lastActivityAt to null when the provider session directory fails", () => Effect.gen(function* () { diff --git a/apps/server/src/mcp/toolkits/sessions/handlers.ts b/apps/server/src/mcp/toolkits/sessions/handlers.ts index bfc7aea609f4..0ea2008bbf73 100644 --- a/apps/server/src/mcp/toolkits/sessions/handlers.ts +++ b/apps/server/src/mcp/toolkits/sessions/handlers.ts @@ -20,6 +20,7 @@ import { SessionOrchestrationWorktreeNotEmptyError, type ServerProvider, type SessionUsageSnapshot, + supersededReportNotice, type SettleSessionInput, type SettleSessionWorktreeOutcome, type SpawnSessionInput, @@ -144,6 +145,15 @@ export const canReadThreadReports = (input: { export const REPORT_NOT_ACCESSIBLE_MESSAGE = "Report not accessible: it does not exist, has not been posted yet, or belongs to a session outside this session's read scope (own spawned sessions and their siblings)."; +// One message for both ways a supersedesReportId can fail to resolve — no +// such report, or a report on another thread. Amending another session's +// report is not a weaker version of amending your own: a report is a +// session's account of its own work, so only the thread that posted one may +// replace it. Saying which of the two went wrong would also turn post_report +// into a probe for which report ids exist elsewhere. +export const SUPERSEDES_REPORT_NOT_FOUND_MESSAGE = + "supersedesReportId does not name a report posted by this session. Pass the reportId returned by your own earlier post_report call on this thread; a report can only be amended by the session that posted it."; + const isHighSurrogate = (code: number) => code >= 0xd800 && code <= 0xdbff; const isLowSurrogate = (code: number) => code >= 0xdc00 && code <= 0xdfff; @@ -210,7 +220,7 @@ export const buildPingSessionSnapshot = (input: { // contract holds across providers without the parent having to remember to // ask for it. post_report is what wakes the parent up. const SPAWNED_SESSION_REPORT_INSTRUCTIONS = - "\n\n---\nYou were spawned by another Phoenix agent session to do the work above. When the work is complete — or you determine it cannot be completed — call the `post_report` tool exactly once with status (success/failure/partial), a concise markdown summary of what you did, and any artifacts (files, branches, PR URLs). If the summary is long, also pass a 1-3 sentence `abstract`. The report is delivered to the session that spawned you."; + "\n\n---\nYou were spawned by another Phoenix agent session to do the work above. When the work is complete — or you determine it cannot be completed — call the `post_report` tool exactly once with status (success/failure/partial), a concise markdown summary of what you did, and any artifacts (files, branches, PR URLs). If the summary is long, also pass a 1-3 sentence `abstract`. The report is delivered to the session that spawned you.\n\nIf you receive a further instruction AFTER you have already posted your report, do the new work and then post an AMENDING report: call `post_report` again with `supersedesReportId` set to the reportId of the report you are replacing. The amended report becomes the record. Never claim in a report that you did something you had not yet done when that report was written — describe what the late instruction was and what you did about it."; // Enough to tell the caller what is at stake without turning a refusal into a // transcript of a large working tree. @@ -1004,6 +1014,21 @@ export const make = Effect.gen(function* () { const postReport = Effect.fn("SessionsToolkit.postReport")(function* (input: PostReportInput) { const scope = yield* requireSessionsCapability; const caller = yield* requireShell(scope.threadId); + + // Resolved before anything is dispatched: an amendment naming a report + // that does not exist (or belongs to another thread) would otherwise + // persist a dangling link that no reader could follow. + if (input.supersedesReportId !== undefined) { + const superseded = yield* reportRepository + .findByReportId({ reportId: input.supersedesReportId }) + .pipe(Effect.mapError(operationError("Failed to read the report being superseded"))); + if (Option.isNone(superseded) || superseded.value.threadId !== scope.threadId) { + return yield* new SessionOrchestrationInvalidInputError({ + message: SUPERSEDES_REPORT_NOT_FOUND_MESSAGE, + }); + } + } + const createdAt = yield* nowIso; const reportId = yield* randomUUID; // Captured now, not agent-supplied: what this session cost by the time @@ -1031,6 +1056,9 @@ export const make = Effect.gen(function* () { ? { completionPercent: input.completionPercent } : {}), usage, + ...(input.supersedesReportId !== undefined + ? { supersedesReportId: input.supersedesReportId } + : {}), createdAt, }), ); @@ -1049,6 +1077,9 @@ export const make = Effect.gen(function* () { ? { completionPercent: input.completionPercent } : {}), usage, + ...(input.supersedesReportId !== undefined + ? { supersedesReportId: input.supersedesReportId } + : {}), // post_report is by definition the agent speaking for itself; only the // reactor's terminal reports are system-origin. origin: "agent" as const, @@ -1149,6 +1180,17 @@ export const make = Effect.gen(function* () { : {}), artifacts: report.artifacts, ...(report.usage !== undefined ? { usage: report.usage } : {}), + ...(report.supersedesReportId !== null + ? { supersedesReportId: report.supersedesReportId } + : {}), + // A caller paging an old body must learn a newer account exists — both + // as an id it can follow and as prose it cannot skim past. + ...(report.supersededByReportId !== undefined + ? { + supersededByReportId: report.supersededByReportId, + supersededNotice: supersededReportNotice(report.supersededByReportId), + } + : {}), createdAt: report.createdAt, }; }); diff --git a/apps/server/src/mcp/toolkits/sessions/tools.ts b/apps/server/src/mcp/toolkits/sessions/tools.ts index d4ff11651bd7..3fe4e46017b8 100644 --- a/apps/server/src/mcp/toolkits/sessions/tools.ts +++ b/apps/server/src/mcp/toolkits/sessions/tools.ts @@ -118,7 +118,7 @@ export const SettleSessionTool = Tool.make("settle_session", { export const ReadReportTool = Tool.make("read_report", { description: - "Read the full body of a completion report posted by a session this session spawned, or by a sibling session (one spawned by the same parent). Pass the reportId from a report envelope, or a threadId to get that thread's latest report. Large reports paginate via offset/maxChars; the result also carries the report's origin (agent vs Phoenix-synthesized) and full findings/validation/recommendation.", + "Read the full body of a completion report posted by a session this session spawned, or by a sibling session (one spawned by the same parent). Pass the reportId from a report envelope, or a threadId to get that thread's latest report. Large reports paginate via offset/maxChars; the result also carries the report's origin (agent vs Phoenix-synthesized) and full findings/validation/recommendation. A report that has been amended comes back with supersededByReportId and a supersededNotice: read that newer report instead, it is the session's current account.", parameters: ReadReportInput, success: ReadReportResult, failure: SessionOrchestrationError, @@ -131,7 +131,7 @@ export const ReadReportTool = Tool.make("read_report", { export const PostReportTool = Tool.make("post_report", { description: - "Post a completion report for THIS session's work: status, a concise markdown summary, and any artifacts (files, branches, PR URLs). For a long summary, also pass a 1-3 sentence abstract: large reports are delivered to the spawning session as a compact envelope, and the abstract is what it sees first. If another session spawned this one, the report is delivered to it automatically; the user also sees the report as a card in this thread. Call once when your assigned work is finished (or clearly failed). Optionally include machine-readable fields: findings (array of {title, severity: info|low|medium|high|critical, detail?}), validation ({performed: string[], gaps: string[]}), recommendation (short string), and completionPercent (0-100). The result also carries a best-effort usage snapshot (tokens, turn count, elapsed time since spawn) captured automatically at post time — this is not something you supply.", + "Post a completion report for THIS session's work: status, a concise markdown summary, and any artifacts (files, branches, PR URLs). For a long summary, also pass a 1-3 sentence abstract: large reports are delivered to the spawning session as a compact envelope, and the abstract is what it sees first. If another session spawned this one, the report is delivered to it automatically; the user also sees the report as a card in this thread. Call once when your assigned work is finished (or clearly failed). If an instruction reaches you AFTER you already reported, do the work and post an AMENDING report: call post_report again with supersedesReportId set to your earlier reportId (it must be a report you posted on this thread, and one that has not itself been superseded — amendments form a single linear chain, so always amend the newest report). The amendment becomes the session's current report and is delivered to the spawning session marked as an amendment; the superseded report stays readable and is flagged as superseded. Never describe work in a report as done when it was not done at the time that report was written. Optionally include machine-readable fields: findings (array of {title, severity: info|low|medium|high|critical, detail?}), validation ({performed: string[], gaps: string[]}), recommendation (short string), and completionPercent (0-100). The result also carries a best-effort usage snapshot (tokens, turn count, elapsed time since spawn) captured automatically at post time — this is not something you supply.", parameters: PostReportInput, success: SessionReport, failure: SessionOrchestrationError, diff --git a/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts b/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts index ca55bafab161..517039097d80 100644 --- a/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts +++ b/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts @@ -357,6 +357,34 @@ describe("formatReportMessage", () => { expect(text).toContain("Phoenix generated a partial report"); }); + it("leads an amending report with what it supersedes", () => { + const text = formatReportMessage( + "Spawned worker", + report({ + supersedesReportId: "report-original", + title: "Also handled the late instruction", + }), + ); + // The parent may already have acted on the superseded report, so the + // amendment has to announce itself before the summary it looks like. + expect(text).toContain("AMENDED report (supersedes report-original)"); + expect(text.indexOf("AMENDED report")).toBeLessThan(text.indexOf("posted a success report")); + }); + + it("keeps the amendment lead on an envelope-sized report", () => { + const summary = "s".repeat(SESSION_REPORT_INLINE_MAX_CHARS + 1); + const text = formatReportMessage( + "Spawned worker", + report({ summary, abstract: "The short version.", supersedesReportId: "report-original" }), + ); + expect(text).toContain("AMENDED report (supersedes report-original)"); + expect(text).toContain('read_report with reportId "report-1"'); + }); + + it("says nothing about amendment for an ordinary report", () => { + expect(formatReportMessage("Spawned worker", report())).not.toContain("AMENDED"); + }); + it("lists artifacts with their labels", () => { const text = formatReportMessage( "Spawned worker", diff --git a/apps/server/src/orchestration/Layers/SessionSpawnReactor.ts b/apps/server/src/orchestration/Layers/SessionSpawnReactor.ts index 3d2c2b7a8aea..0a1d3661552e 100644 --- a/apps/server/src/orchestration/Layers/SessionSpawnReactor.ts +++ b/apps/server/src/orchestration/Layers/SessionSpawnReactor.ts @@ -76,12 +76,19 @@ export const formatReportMessage = (childTitle: string, report: SessionReport): : `- ${artifact.kind}: ${artifact.value}`, ) .join("\n")}`; + // An amendment leads with the fact that it replaces an earlier report: a + // parent that already acted on the superseded one has to see that first, + // before it reads a summary it thinks it has seen. + const amendment = + report.supersedesReportId !== undefined + ? `AMENDED report (supersedes ${report.supersedesReportId}). ` + : ""; // A synthesized report must never read as if the child wrote it: the parent // decides what to do next based on who is claiming the work is over. const lead = report.origin === "system" - ? `[Phoenix] Spawned session "${childTitle}" ended without posting a report. Phoenix generated a ${report.status} report for it: ${report.title}` - : `[Phoenix] Spawned session "${childTitle}" posted a ${report.status} report: ${report.title}`; + ? `[Phoenix] ${amendment}Spawned session "${childTitle}" ended without posting a report. Phoenix generated a ${report.status} report for it: ${report.title}` + : `[Phoenix] ${amendment}Spawned session "${childTitle}" posted a ${report.status} report: ${report.title}`; // Reports at or under the inline threshold are delivered whole — agent and // system reports alike. Larger ones become a compact envelope: abstract // plus addressing, with the body (and full findings/validation) behind From fd4f3ff355fa5d5dfdd8b11140fa846212977ccc Mon Sep 17 00:00:00 2001 From: roughcoder Date: Wed, 12 Aug 2026 15:35:36 +0100 Subject: [PATCH 4/5] docs(internals): record how report amendment works Co-Authored-By: Claude Opus 5 (1M context) --- docs/internals/session-orchestration.md | 42 +++++++++++++++++++++++-- 1 file changed, 40 insertions(+), 2 deletions(-) diff --git a/docs/internals/session-orchestration.md b/docs/internals/session-orchestration.md index 15b29254bf3a..d57ad5a64fdf 100644 --- a/docs/internals/session-orchestration.md +++ b/docs/internals/session-orchestration.md @@ -51,8 +51,9 @@ agent session ── MCP tool call ──> apps/server/src/mcp/toolkits/sessions after confirming there is no live provider binding. - **Persistence** — migrations 041 (`projection_threads.spawned_by_thread_id`), 042 (`projection_thread_reports`), 043 (structured report fields), 044 (session stop audit), 045 - (`projection_thread_reports.origin`), and 046 (`projection_thread_reports.abstract`), with - hydration through `ProjectionPipeline` and `ProjectionSnapshotQuery`. + (`projection_thread_reports.origin`), 046 (`projection_thread_reports.abstract`), and 048 + (`projection_thread_reports.supersedes_report_id`), with hydration through `ProjectionPipeline` + and `ProjectionSnapshotQuery`. - **Reactor** — `apps/server/src/orchestration/Layers/SessionSpawnReactor.ts` watches `thread.report-posted` and `thread.session-set` domain events. A report on a thread with `spawnedByThreadId` starts a turn on the parent carrying the report. Turn injection is the wake @@ -120,6 +121,43 @@ rides along automatically to `SessionReportEnvelope` and `read_report`. Deliberately no cost estimate anywhere in this: provider price tables go stale, so tokens are the stable currency and converting to cost, if a caller wants that, is a client-side concern. +## Amending a report + +A report is a claim about work, and a queued instruction can arrive after the child already made +it. Left alone, the stale report stays the record — and the incident that motivated this was worse +than stale: the child, having reported, then claimed compliance with an instruction it had never +acted on. So `post_report` takes an optional `supersedesReportId`, and the amendment — not the +original — becomes the session's current account. + +Nothing is rewritten. The projection is append-only: the amendment stores a forward link +(`supersedes_report_id`), the superseded report keeps its own row, its own event, and its own body, +and the reverse link (`supersededByReportId`) is _derived_ on every read path by asking which later +report points back at this one. That is why there is no "superseded" flag column to keep in sync, +and why a report can be read from either end of the chain. + +The reference is validated in the toolkit before anything is dispatched: it must name a report on +the _calling thread_. A report is a session's account of its own work, so amending another +session's report is not a weaker case of amending your own — it is refused, with the same message +as an unknown id so the denial cannot double as a probe for which report ids exist elsewhere. A +dangling link would be worse than a rejection: no reader could follow it. + +Both ends travel outward. Envelopes and `read_session` carry `supersedesReportId` / +`supersededByReportId`; the parent notification for an amending report leads with +`AMENDED report (supersedes …)` before the summary, because a parent that already acted on the +superseded report has to see that first. `read_report` on a superseded report returns the body it +always did — the record is append-only — plus `supersededByReportId` and a `supersededNotice` +sentence: a caller paging an old report must learn a newer one exists, and cannot be relied on to +notice a field it was not looking for. + +`SPAWNED_SESSION_REPORT_INSTRUCTIONS` (force-appended to every spawned prompt) tells children the +rule directly: an instruction arriving after you reported means post an amending report — never +claim retroactive compliance. + +The event payload change follows the event-sourcing rule: `supersedesReportId` is optional on both +the `thread.report.post` command and the `thread.report-posted` payload, so every already-persisted +report event replays unchanged. `supersededByReportId` is deliberately _not_ in the payload — at +post time no such report exists yet, and inventing one would make the event a lie. + ## Settling a child Sessions do not settle themselves — a finished report is not the same claim as "this thread is From ff81ae552a171f5ba0b3a5d6bea919a9077b46f4 Mon Sep 17 00:00:00 2001 From: roughcoder Date: Wed, 12 Aug 2026 17:34:37 +0100 Subject: [PATCH 5/5] fix(sessions): keep amendment chains linear, enforced in the decider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review found A→B and A→C was possible, leaving reverse navigation and latest-report selection free to disagree about which report is current. Superseding an already-superseded report is now refused and the caller is handed the head of the chain, which also settles the concurrent-fork race: the second writer loses with an error naming where to re-attach. The authoritative check lives in the decider, not the toolkit. Handler-level validation cannot prevent a fork — two amendments can both pass their pre-checks before either dispatches — whereas the decider runs against the folded read model serialized with command processing, and also covers internally dispatched report posts that never reach the toolkit. The toolkit keeps a pre-check purely for error quality, and re-reads the chain when a dispatch is rejected so a race loser gets the same structured error rather than a generic dispatch failure. Both sides share one implementation and one wording so they cannot disagree. Tests: fork rejection including the race shape, chain-head reporting through a longer chain, a one-turn reactor run asserting AMENDED delivery reaches the parent (not just the formatter), and synthesized-report resurrection — a Phoenix terminal report superseded by the agent's own account after resume. Co-Authored-By: Claude Opus 5 (1M context) --- .../mcp/toolkits/sessions/handlers.test.ts | 196 ++++++++++++++++-- .../src/mcp/toolkits/sessions/handlers.ts | 132 +++++++++--- .../Layers/SessionSpawnReactor.test.ts | 84 ++++++++ .../src/orchestration/decider.reports.test.ts | 166 ++++++++++++++- apps/server/src/orchestration/decider.ts | 34 ++- .../Layers/ProjectionRepositories.test.ts | 43 ++++ docs/internals/session-orchestration.md | 32 ++- .../src/sessionOrchestration.test.ts | 77 +++++++ .../contracts/src/sessionOrchestration.ts | 102 +++++++++ 9 files changed, 812 insertions(+), 54 deletions(-) diff --git a/apps/server/src/mcp/toolkits/sessions/handlers.test.ts b/apps/server/src/mcp/toolkits/sessions/handlers.test.ts index 360b12031fc0..a34b4d625377 100644 --- a/apps/server/src/mcp/toolkits/sessions/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/sessions/handlers.test.ts @@ -11,6 +11,7 @@ import { SESSION_REPORT_INLINE_MAX_CHARS, SessionOrchestrationDeniedError, SessionOrchestrationInvalidInputError, + SessionOrchestrationReportAlreadySupersededError, type SessionReport, type SessionUsageSnapshot, supersededReportNotice, @@ -28,6 +29,7 @@ import * as GitWorkflowService from "../../../git/GitWorkflowService.ts"; import * as OrchestrationEngine from "../../../orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; import * as ThreadTurnBootstrap from "../../../orchestration/ThreadTurnBootstrap.ts"; +import { OrchestrationCommandInvariantError } from "../../../orchestration/Errors.ts"; import { PersistenceSqlError } from "../../../persistence/Errors.ts"; import { type ProjectionThreadReport, @@ -907,15 +909,53 @@ const projectedReport = ( }); describe("post_report supersession (handler)", () => { + // These mocks are annotated against the real service shapes rather than + // left to inference: an untyped mock widens the handler's error channel to + // `unknown`, which silently defeats the `Effect.flip` assertions below. + type WritePathOverrides = { + readonly dispatch: OrchestrationEngine.OrchestrationEngineShape["dispatch"]; + readonly enqueueCommand: ServerRuntimeStartup.ServerRuntimeStartup["Service"]["enqueueCommand"]; + }; + // Captures what post_report dispatched, so the tests can assert the // amendment link reached the command rather than only the tool result. - const capturingDispatch = (captured: Array) => ({ - dispatch: (command: OrchestrationCommand) => + const capturingDispatch = (captured: Array): WritePathOverrides => ({ + dispatch: (command) => Effect.sync(() => { captured.push(command); return { sequence: 1 }; }), - enqueueCommand: (effect: Effect.Effect) => effect, + enqueueCommand: (effect) => effect, + }); + + // The decider's actual refusal, so the race tests fail the way the real + // dispatch path fails rather than with a stand-in Error. + const decliningDispatch = ( + detail: string, + onDispatch: () => void = () => {}, + ): WritePathOverrides => ({ + dispatch: () => + Effect.sync(onDispatch).pipe( + Effect.andThen( + Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: "thread.report.post", + detail, + }), + ), + ), + ), + enqueueCommand: (effect) => effect, + }); + + // Faithful to the real query, which is scoped by thread: a report on + // another thread is simply not in the result, which is exactly why the + // cross-thread case cannot be distinguished from an unknown id. + const threadReports = ( + reports: ReadonlyArray, + ): Pick => ({ + listByThreadId: ({ threadId }) => + Effect.succeed(reports.filter((entry) => entry.threadId === threadId)), }); it.effect("carries a valid supersedesReportId into the command and the result", () => @@ -931,8 +971,7 @@ describe("post_report supersession (handler)", () => { }), { ...capturingDispatch(captured), - findByReportId: () => - Effect.succeed(Option.some(projectedReport({ reportId: "report-original" }))), + ...threadReports([projectedReport({ reportId: "report-original" })]), }, ); @@ -958,7 +997,7 @@ describe("post_report supersession (handler)", () => { }), { ...capturingDispatch(captured), - findByReportId: () => Effect.succeed(Option.none()), + ...threadReports([projectedReport({ reportId: "report-original" })]), }, ).pipe(Effect.flip); @@ -982,15 +1021,9 @@ describe("post_report supersession (handler)", () => { }), { ...capturingDispatch(captured), - findByReportId: () => - Effect.succeed( - Option.some( - projectedReport({ - reportId: "report-of-another-thread", - threadId: childThreadId, - }), - ), - ), + ...threadReports([ + projectedReport({ reportId: "report-of-another-thread", threadId: childThreadId }), + ]), }, ).pipe(Effect.flip); @@ -1003,10 +1036,141 @@ describe("post_report supersession (handler)", () => { }), ); + it.effect("refuses to fork a chain, naming the report to supersede instead", () => + Effect.gen(function* () { + const captured: Array = []; + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Second amendment of the same report", + summary: "Would fork the chain.", + supersedesReportId: "report-original", + }), + { + ...capturingDispatch(captured), + ...threadReports([ + projectedReport({ reportId: "report-original" }), + projectedReport({ + reportId: "report-amendment", + supersedesReportId: "report-original", + }), + ]), + }, + ).pipe(Effect.flip); + + expect(error).toBeInstanceOf(SessionOrchestrationReportAlreadySupersededError); + expect(error).toMatchObject({ + reportId: "report-original", + supersededByReportId: "report-amendment", + chainHeadReportId: "report-amendment", + }); + expect(captured).toHaveLength(0); + }), + ); + + it.effect("points a refused fork at the head of a longer chain", () => + Effect.gen(function* () { + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Amendment of a stale link", + summary: "Two amendments behind.", + supersedesReportId: "report-original", + }), + { + ...capturingDispatch([]), + ...threadReports([ + projectedReport({ reportId: "report-original" }), + projectedReport({ reportId: "report-second", supersedesReportId: "report-original" }), + projectedReport({ reportId: "report-third", supersedesReportId: "report-second" }), + ]), + }, + ).pipe(Effect.flip); + + expect(error).toMatchObject({ + supersededByReportId: "report-second", + chainHeadReportId: "report-third", + }); + expect(error.message).toContain("supersede report-third instead"); + }), + ); + + it.effect("turns a lost amendment race into the same actionable error", () => + Effect.gen(function* () { + // The race the decider resolves: this caller's pre-check passed, another + // amendment took the chain head, and the decider rejected the dispatch. + // A generic dispatch failure would leave the loser with nowhere to + // re-attach, so the chain is re-read and reported instead. + let dispatched = false; + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Lost the race", + summary: "Superseded concurrently.", + supersedesReportId: "report-original", + }), + { + ...decliningDispatch( + "Report report-original is already superseded by report-winner", + () => { + dispatched = true; + }, + ), + // First read (pre-check) sees an unsuperseded report; the re-read + // after the rejection sees the winner. + listByThreadId: () => + Effect.succeed( + dispatched + ? [ + projectedReport({ reportId: "report-original" }), + projectedReport({ + reportId: "report-winner", + supersedesReportId: "report-original", + }), + ] + : [projectedReport({ reportId: "report-original" })], + ), + }, + ).pipe(Effect.flip); + + expect(error).toBeInstanceOf(SessionOrchestrationReportAlreadySupersededError); + expect(error).toMatchObject({ + reportId: "report-original", + supersededByReportId: "report-winner", + chainHeadReportId: "report-winner", + }); + }), + ); + + it.effect("lets the original dispatch failure stand when the chain did not move", () => + Effect.gen(function* () { + const error = yield* runHandler( + (handlers) => + handlers.post_report({ + status: "success", + title: "Dispatch broke", + summary: "Unrelated failure.", + supersedesReportId: "report-original", + }), + { + // A failure that has nothing to do with supersession. + ...decliningDispatch("Projection write failed while persisting the report."), + listByThreadId: () => Effect.succeed([projectedReport({ reportId: "report-original" })]), + }, + ).pipe(Effect.flip); + + // Not relabelled as a supersession problem: it was not one. + expect(error).not.toBeInstanceOf(SessionOrchestrationReportAlreadySupersededError); + }), + ); + it.effect("posts an ordinary report without touching the report repository", () => Effect.gen(function* () { const captured: Array = []; - // findByReportId is left unmocked here, so it dies if called: a report + // listByThreadId is left unmocked here, so it dies if called: a report // with no supersedesReportId must not pay for a lookup. const result = yield* runHandler( (handlers) => diff --git a/apps/server/src/mcp/toolkits/sessions/handlers.ts b/apps/server/src/mcp/toolkits/sessions/handlers.ts index 0ea2008bbf73..4adb208a8647 100644 --- a/apps/server/src/mcp/toolkits/sessions/handlers.ts +++ b/apps/server/src/mcp/toolkits/sessions/handlers.ts @@ -1,4 +1,5 @@ import { + checkReportSupersession, CommandId, isProviderAvailable, MessageId, @@ -10,12 +11,14 @@ import { READ_REPORT_MAX_CHARS, type ReadReportInput, type ReadSessionResult, + reportAlreadySupersededMessage, type RuntimeMode, SESSION_SPAWN_MAX_CHILDREN, SESSION_SPAWN_MAX_DEPTH, SessionOrchestrationDeniedError, SessionOrchestrationInvalidInputError, SessionOrchestrationOperationError, + SessionOrchestrationReportAlreadySupersededError, SessionOrchestrationUnavailableError, SessionOrchestrationWorktreeNotEmptyError, type ServerProvider, @@ -1015,20 +1018,85 @@ export const make = Effect.gen(function* () { const scope = yield* requireSessionsCapability; const caller = yield* requireShell(scope.threadId); - // Resolved before anything is dispatched: an amendment naming a report - // that does not exist (or belongs to another thread) would otherwise - // persist a dangling link that no reader could follow. - if (input.supersedesReportId !== undefined) { - const superseded = yield* reportRepository - .findByReportId({ reportId: input.supersedesReportId }) - .pipe(Effect.mapError(operationError("Failed to read the report being superseded"))); - if (Option.isNone(superseded) || superseded.value.threadId !== scope.threadId) { + // Friendly pre-check. The decider runs the same check against the folded + // read model and is the authority — this one exists so the common case + // fails with a specific, structured error instead of a dispatch failure. + // Reading the whole thread's reports (rather than one row) is what makes + // the chain-head answer available. + const supersedesReportId = input.supersedesReportId; + if (supersedesReportId !== undefined) { + const reports = yield* reportRepository + .listByThreadId({ threadId: scope.threadId }) + .pipe(Effect.mapError(operationError("Failed to read this session's reports"))); + const check = checkReportSupersession(reports, supersedesReportId); + if (check._tag === "unknown-report") { return yield* new SessionOrchestrationInvalidInputError({ message: SUPERSEDES_REPORT_NOT_FOUND_MESSAGE, }); } + if (check._tag === "already-superseded") { + return yield* new SessionOrchestrationReportAlreadySupersededError({ + message: reportAlreadySupersededMessage({ + reportId: supersedesReportId, + supersededByReportId: check.supersededByReportId, + chainHeadReportId: check.chainHeadReportId, + }), + reportId: supersedesReportId, + supersededByReportId: check.supersededByReportId, + chainHeadReportId: check.chainHeadReportId, + }); + } } + /** + * Turn a lost amendment race into the same actionable error the pre-check + * would have given. + * + * Between the pre-check and the decider, another amendment can take the + * chain head. The decider rejects this command — correctly — but through + * dispatch that surfaces as a generic operation failure, which tells the + * caller nothing about where to re-attach. So on failure, re-read the + * chain: if it moved, report that; otherwise the dispatch failed for some + * other reason and that error stands. + */ + const withSupersessionRaceDetail = ( + effect: Effect.Effect<{ readonly sequence: number }, SessionOrchestrationOperationError>, + ): Effect.Effect< + { readonly sequence: number }, + SessionOrchestrationOperationError | SessionOrchestrationReportAlreadySupersededError + > => + supersedesReportId === undefined + ? effect + : effect.pipe( + Effect.catch((dispatchError) => + reportRepository.listByThreadId({ threadId: scope.threadId }).pipe( + // The recheck is diagnostic only; if it fails, the original + // dispatch error is still the truthful thing to report. + Effect.catch(() => Effect.succeed>([])), + Effect.flatMap((reports) => { + const check = checkReportSupersession(reports, supersedesReportId); + return check._tag === "already-superseded" + ? Effect.fail< + | SessionOrchestrationOperationError + | SessionOrchestrationReportAlreadySupersededError + >( + new SessionOrchestrationReportAlreadySupersededError({ + message: reportAlreadySupersededMessage({ + reportId: supersedesReportId, + supersededByReportId: check.supersededByReportId, + chainHeadReportId: check.chainHeadReportId, + }), + reportId: supersedesReportId, + supersededByReportId: check.supersededByReportId, + chainHeadReportId: check.chainHeadReportId, + }), + ) + : Effect.fail(dispatchError); + }), + ), + ), + ); + const createdAt = yield* nowIso; const reportId = yield* randomUUID; // Captured now, not agent-supplied: what this session cost by the time @@ -1038,29 +1106,31 @@ export const make = Effect.gen(function* () { createdAt: caller.createdAt, latestTurn: caller.latestTurn, }); - yield* enqueue( - engine.dispatch({ - type: "thread.report.post", - commandId: yield* serverCommandId("mcp-post-report"), - threadId: scope.threadId, - reportId, - status: input.status, - title: input.title, - summary: input.summary, - ...(input.abstract !== undefined ? { abstract: input.abstract } : {}), - artifacts: input.artifacts ?? [], - ...(input.findings !== undefined ? { findings: input.findings } : {}), - ...(input.validation !== undefined ? { validation: input.validation } : {}), - ...(input.recommendation !== undefined ? { recommendation: input.recommendation } : {}), - ...(input.completionPercent !== undefined - ? { completionPercent: input.completionPercent } - : {}), - usage, - ...(input.supersedesReportId !== undefined - ? { supersedesReportId: input.supersedesReportId } - : {}), - createdAt, - }), + yield* withSupersessionRaceDetail( + enqueue( + engine.dispatch({ + type: "thread.report.post", + commandId: yield* serverCommandId("mcp-post-report"), + threadId: scope.threadId, + reportId, + status: input.status, + title: input.title, + summary: input.summary, + ...(input.abstract !== undefined ? { abstract: input.abstract } : {}), + artifacts: input.artifacts ?? [], + ...(input.findings !== undefined ? { findings: input.findings } : {}), + ...(input.validation !== undefined ? { validation: input.validation } : {}), + ...(input.recommendation !== undefined ? { recommendation: input.recommendation } : {}), + ...(input.completionPercent !== undefined + ? { completionPercent: input.completionPercent } + : {}), + usage, + ...(input.supersedesReportId !== undefined + ? { supersedesReportId: input.supersedesReportId } + : {}), + createdAt, + }), + ), ); return { reportId, diff --git a/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts b/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts index 517039097d80..fe98004e4ad0 100644 --- a/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts +++ b/apps/server/src/orchestration/Layers/SessionSpawnReactor.test.ts @@ -443,6 +443,90 @@ describe("formatReportMessage", () => { }); }); +const reportPostedEvent = ( + posted: ReturnType, + sequence: number, +): OrchestrationEvent => ({ + sequence, + eventId: EventId.make(`event-report-${posted.reportId}`), + aggregateKind: "thread", + aggregateId: CHILD_ID, + occurredAt: CREATED_AT, + commandId: CommandId.make(`command-report-${posted.reportId}`), + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.report-posted", + payload: { + threadId: CHILD_ID, + report: { ...posted, threadId: CHILD_ID }, + updatedAt: CREATED_AT, + }, +}); + +// Drives the real reactor over a real thread.report-posted event, rather than +// calling the formatter directly: what a parent actually receives depends on +// the reactor resolving the child, finding its spawner, and dispatching a turn +// start — none of which formatter tests exercise. +describe("SessionSpawnReactor amended report delivery", () => { + const deliveredToParent = (commands: ReadonlyArray) => + commands + .filter((command) => command.type === "thread.turn.start") + .filter((command) => command.threadId === PARENT_ID) + .map((command) => (command.type === "thread.turn.start" ? command.message.text : "")); + + it.effect("delivers an amending report to the parent led by what it supersedes", () => + Effect.scoped( + createHarness({ + status: "ready", + queued: [], + boundaryEvents: [ + reportPostedEvent( + report({ + reportId: "report-amendment", + title: "Also handled the late instruction", + summary: "The queued instruction arrived after the first report; it is done now.", + supersedesReportId: "report-original", + }), + 2, + ), + ], + }).pipe( + Effect.map(({ commands }) => { + const delivered = deliveredToParent(commands); + expect(delivered).toHaveLength(1); + const text = delivered[0] ?? ""; + expect(text).toContain("AMENDED report (supersedes report-original)"); + // The amendment marker precedes the summary the parent may think it + // has already read. + expect(text.indexOf("AMENDED report")).toBeLessThan( + text.indexOf("The queued instruction arrived"), + ); + expect(text).toContain("(spawned thread: child-thread)"); + }), + Effect.provide(NodeServices.layer), + ), + ), + ); + + it.effect("delivers an ordinary report with no amendment marker", () => + Effect.scoped( + createHarness({ + status: "ready", + queued: [], + boundaryEvents: [reportPostedEvent(report({ reportId: "report-original" }), 2)], + }).pipe( + Effect.map(({ commands }) => { + const delivered = deliveredToParent(commands); + expect(delivered).toHaveLength(1); + expect(delivered[0]).not.toContain("AMENDED"); + }), + Effect.provide(NodeServices.layer), + ), + ), + ); +}); + describe("buildTerminalReportSummary", () => { it("says the session was stopped and flags the work as unfinished", () => { const summary = buildTerminalReportSummary({ diff --git a/apps/server/src/orchestration/decider.reports.test.ts b/apps/server/src/orchestration/decider.reports.test.ts index 44693274debb..50729cbe49c2 100644 --- a/apps/server/src/orchestration/decider.reports.test.ts +++ b/apps/server/src/orchestration/decider.reports.test.ts @@ -1,5 +1,6 @@ import { CommandId, + type OrchestrationCommand, ProjectId, ProviderInstanceId, ThreadId, @@ -10,10 +11,19 @@ import { expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import { decideOrchestrationCommand } from "./decider.ts"; +import type { OrchestrationCommandInvariantError } from "./Errors.ts"; const CREATED_AT = "2026-01-01T00:00:00.000Z"; const POSTED_AT = "2026-01-01T01:00:00.000Z"; +// decideOrchestrationCommand can also fail with a PlatformError, so the +// flipped error is a union; narrow before reading the invariant's detail +// rather than asserting against a field the type does not guarantee. +const invariantDetail = (error: { readonly _tag: string }): string => + error._tag === "OrchestrationCommandInvariantError" + ? (error as OrchestrationCommandInvariantError).detail + : `expected an OrchestrationCommandInvariantError, received ${error._tag}`; + const readModel: OrchestrationReadModel = { snapshotSequence: 0, projects: [], @@ -182,7 +192,9 @@ it.layer(NodeServices.layer)("report decider", (it) => { supersedesReportId: "report-1", createdAt: POSTED_AT, }, - readModel, + // The superseded report has to exist on the thread: the decider + // refuses a dangling amendment link. + readModel: readModelWithReports([{ reportId: "report-1" }]), }); const event = Array.isArray(result) ? result[0] : result; @@ -196,6 +208,158 @@ it.layer(NodeServices.layer)("report decider", (it) => { }), ); + // A thread that already carries reports, so the decider's supersession + // check has a folded read model to work against — the same shape command + // processing sees, which is what makes this check race-proof. + const readModelWithReports = ( + reports: ReadonlyArray<{ readonly reportId: string; readonly supersedesReportId?: string }>, + ): OrchestrationReadModel => ({ + ...readModel, + threads: readModel.threads.map((thread) => ({ + ...thread, + reports: reports.map((entry) => ({ + reportId: entry.reportId, + threadId: ThreadId.make("thread-1"), + status: "success" as const, + title: "Did the work", + summary: "All done.", + artifacts: [], + origin: "agent" as const, + ...(entry.supersedesReportId !== undefined + ? { supersedesReportId: entry.supersedesReportId } + : {}), + createdAt: POSTED_AT, + })), + })), + }); + + const amendmentCommand = (input: { + readonly reportId: string; + readonly supersedesReportId: string; + }) => + ({ + type: "thread.report.post", + commandId: CommandId.make(`cmd-${input.reportId}`), + threadId: ThreadId.make("thread-1"), + reportId: input.reportId, + status: "success", + title: "Amended", + summary: "Amended account.", + artifacts: [], + supersedesReportId: input.supersedesReportId, + createdAt: POSTED_AT, + }) satisfies Extract; + + it.effect("accepts an amendment of the newest report in a chain", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: amendmentCommand({ reportId: "report-c", supersedesReportId: "report-b" }), + readModel: readModelWithReports([ + { reportId: "report-a" }, + { reportId: "report-b", supersedesReportId: "report-a" }, + ]), + }); + const event = Array.isArray(result) ? result[0] : result; + + expect(event.type).toBe("thread.report-posted"); + if (event.type === "thread.report-posted") { + expect(event.payload.report.supersedesReportId).toBe("report-b"); + } + }), + ); + + it.effect("refuses to fork a chain by superseding an already-superseded report", () => + Effect.gen(function* () { + // The race, resolved: two amendments of report-a reach the decider; the + // first produced report-b, so the second is rejected here rather than + // creating an a→b / a→c fork that no reader could order. + const result = yield* decideOrchestrationCommand({ + command: amendmentCommand({ reportId: "report-c", supersedesReportId: "report-a" }), + readModel: readModelWithReports([ + { reportId: "report-a" }, + { reportId: "report-b", supersedesReportId: "report-a" }, + ]), + }).pipe(Effect.flip); + + expect(result._tag).toBe("OrchestrationCommandInvariantError"); + // Actionable: names the winner and where to re-attach. + expect(invariantDetail(result)).toContain("report-a is already superseded by report-b"); + expect(invariantDetail(result)).toContain("supersede report-b instead"); + }), + ); + + it.effect("points a rejected amendment at the head of a longer chain", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: amendmentCommand({ reportId: "report-d", supersedesReportId: "report-a" }), + readModel: readModelWithReports([ + { reportId: "report-a" }, + { reportId: "report-b", supersedesReportId: "report-a" }, + { reportId: "report-c", supersedesReportId: "report-b" }, + ]), + }).pipe(Effect.flip); + + expect(result._tag).toBe("OrchestrationCommandInvariantError"); + expect(invariantDetail(result)).toContain("current head of that chain is report-c"); + expect(invariantDetail(result)).toContain("supersede report-c instead"); + }), + ); + + it.effect("refuses an amendment of a report that does not exist on the thread", () => + Effect.gen(function* () { + // The decider is authoritative, so this holds for internally dispatched + // report posts too — not only ones the toolkit pre-checked. + const result = yield* decideOrchestrationCommand({ + command: amendmentCommand({ reportId: "report-b", supersedesReportId: "report-missing" }), + readModel: readModelWithReports([{ reportId: "report-a" }]), + }).pipe(Effect.flip); + + expect(result._tag).toBe("OrchestrationCommandInvariantError"); + expect(invariantDetail(result)).toContain("report-missing"); + }), + ); + + it.effect("lets a real agent report supersede a Phoenix-synthesized one", () => + Effect.gen(function* () { + // Resurrection: the session was stopped, Phoenix synthesized a terminal + // report, then the session resumed and its agent finished the work. The + // agent's account must be able to replace the synthesized one. + const result = yield* decideOrchestrationCommand({ + command: amendmentCommand({ + reportId: "report-agent", + supersedesReportId: "report-synthetic", + }), + readModel: { + ...readModel, + threads: readModel.threads.map((thread) => ({ + ...thread, + reports: [ + { + reportId: "report-synthetic", + threadId: ThreadId.make("thread-1"), + status: "partial" as const, + title: "Session stopped before reporting", + summary: "Phoenix generated this report.", + artifacts: [], + origin: "system" as const, + createdAt: CREATED_AT, + }, + ], + })), + }, + }); + const event = Array.isArray(result) ? result[0] : result; + + expect(event.type).toBe("thread.report-posted"); + if (event.type === "thread.report-posted") { + expect(event.payload.report.supersedesReportId).toBe("report-synthetic"); + // The amendment is the agent speaking; superseding a system report + // does not inherit its origin. + expect(event.payload.report.origin).toBe("agent"); + } + }), + ); + it.effect("rejects a report for an unknown thread", () => Effect.gen(function* () { const result = yield* decideOrchestrationCommand({ diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index a5e529fed352..ec8d90a2d6ee 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1,8 +1,10 @@ import { + checkReportSupersession, EventId, type OrchestrationCommand, type OrchestrationEvent, type OrchestrationReadModel, + reportAlreadySupersededMessage, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; @@ -1461,11 +1463,41 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.report.post": { - yield* requireThread({ + const reportThread = yield* requireThread({ readModel, command, threadId: command.threadId, }); + // Authoritative supersession check. The toolkit runs the same check + // first for a friendlier error, but only this one is serialized with + // command processing: two amendments of the same report racing through + // the toolkit both pass their pre-checks, and this is where the second + // loses. It also covers internally dispatched thread.report.post + // commands, which never pass through the toolkit at all. + // + // Correctness here depends on `reports` being the thread's COMPLETE + // list. The projector caps messages/checkpoints/activities but + // deliberately does not cap reports; capping them would start rejecting + // legitimate amendments of reports that had aged out, as "unknown". + if (command.supersedesReportId !== undefined) { + const check = checkReportSupersession(reportThread.reports, command.supersedesReportId); + if (check._tag === "unknown-report") { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Report '${command.supersedesReportId}' does not exist on thread '${command.threadId}', so it cannot be superseded.`, + }); + } + if (check._tag === "already-superseded") { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: reportAlreadySupersededMessage({ + reportId: command.supersedesReportId, + supersededByReportId: check.supersededByReportId, + chainHeadReportId: check.chainHeadReportId, + }), + }); + } + } return { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index fa323b54deac..6b426b12f1e2 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -277,6 +277,49 @@ projectionRepositoriesLayer("Projection repositories", (it) => { }), ); + it.effect("lets a resumed session's agent report supersede a synthesized one", () => + Effect.gen(function* () { + const reports = yield* ProjectionThreadReportRepository; + const threadId = ThreadId.make("thread-resurrected"); + + // The session was stopped, so Phoenix wrote a terminal report for it. + yield* reports.upsert( + report({ + reportId: "report-synthetic", + threadId, + status: "partial", + title: "Session stopped before reporting", + summary: "Phoenix generated this report.", + origin: "system", + createdAt: "2026-08-12T01:00:00.000Z", + }), + ); + // It came back and finished the work; its own account supersedes the + // stand-in. + yield* reports.upsert( + report({ + reportId: "report-agent", + threadId, + summary: "Resumed and finished the work.", + supersedesReportId: "report-synthetic", + createdAt: "2026-08-12T02:00:00.000Z", + }), + ); + + const synthetic = Option.getOrNull( + yield* reports.findByReportId({ reportId: "report-synthetic" }), + ); + // Still system-origin and still readable — the parent can see both that + // Phoenix stood in and that the agent later spoke for itself. + assert.strictEqual(synthetic?.origin, "system"); + assert.strictEqual(synthetic?.supersededByReportId, "report-agent"); + + const listed = yield* reports.listByThreadId({ threadId }); + assert.strictEqual(listed.at(-1)?.reportId, "report-agent"); + assert.strictEqual(listed.at(-1)?.origin, "agent"); + }), + ); + it.effect("leaves an unamended report with no supersession links", () => Effect.gen(function* () { const reports = yield* ProjectionThreadReportRepository; diff --git a/docs/internals/session-orchestration.md b/docs/internals/session-orchestration.md index d57ad5a64fdf..120f2e196806 100644 --- a/docs/internals/session-orchestration.md +++ b/docs/internals/session-orchestration.md @@ -135,11 +135,33 @@ and the reverse link (`supersededByReportId`) is _derived_ on every read path by report points back at this one. That is why there is no "superseded" flag column to keep in sync, and why a report can be read from either end of the chain. -The reference is validated in the toolkit before anything is dispatched: it must name a report on -the _calling thread_. A report is a session's account of its own work, so amending another -session's report is not a weaker case of amending your own — it is refused, with the same message -as an unknown id so the denial cannot double as a probe for which report ids exist elsewhere. A -dangling link would be worse than a rejection: no reader could follow it. +Chains are **linear, never forked**: superseding a report that is already superseded is refused. +Two reports both amending A would leave "which is current" ambiguous — reverse navigation from A +could reach either, while latest-report selection picks by recency, and the two answers need not +agree. Rather than teach every reader a merge rule, the loser is refused and handed the head of the +chain (`SessionOrchestrationReportAlreadySupersededError` carries `supersededByReportId` and +`chainHeadReportId`, because the caller's next move is mechanical: re-post against the head). + +The **decider** is where that check is authoritative, not the toolkit. Handler-level validation +alone cannot prevent a fork: two amendments of the same report can both pass their pre-checks +before either is dispatched. The decider runs against the folded read model _serialized with +command processing_, so the second one loses deterministically — and it also covers +`thread.report.post` commands dispatched internally, which never pass through the toolkit at all. +The toolkit keeps a pre-check purely for error quality: the common case fails with a structured +error instead of a dispatch failure, and a caller that loses the race gets that same structured +error because post_report re-reads the chain before surfacing the rejection. Both sides share one +implementation (`checkReportSupersession`) and one wording, so they cannot disagree. + +The reference must also name a report on the _calling thread_. A report is a session's account of +its own work, so amending another session's report is not a weaker case of amending your own — it +is refused, with the same message as an unknown id so the denial cannot double as a probe for which +report ids exist elsewhere. A dangling link would be worse than a rejection: no reader could follow +it. + +A Phoenix-synthesized terminal report is amendable like any other, which is what makes resurrection +work: a session that was stopped, had a report synthesized for it, then resumed and finished can +supersede the stand-in with its own account. The superseded row keeps `origin: "system"`, so the +history still shows that Phoenix stood in. Both ends travel outward. Envelopes and `read_session` carry `supersedesReportId` / `supersededByReportId`; the parent notification for an amending report leads with diff --git a/packages/contracts/src/sessionOrchestration.test.ts b/packages/contracts/src/sessionOrchestration.test.ts index a98b82eae94d..94d299b04c59 100644 --- a/packages/contracts/src/sessionOrchestration.test.ts +++ b/packages/contracts/src/sessionOrchestration.test.ts @@ -2,11 +2,88 @@ import { describe, expect, it } from "vite-plus/test"; import * as Schema from "effect/Schema"; import { + checkReportSupersession, ReadSessionResult, + reportAlreadySupersededMessage, SpawnSessionInput, SpawnSessionResult, } from "./sessionOrchestration.ts"; +describe("checkReportSupersession", () => { + it("accepts amending the newest report in a chain", () => { + expect( + checkReportSupersession([{ reportId: "a" }, { reportId: "b", supersedesReportId: "a" }], "b"), + ).toEqual({ _tag: "ok" }); + }); + + it("accepts amending an unsuperseded report that is not the newest row", () => { + // The rule bans forks, not amending an older independent report: `b` here + // supersedes nothing, so `a` still has a single, unambiguous successor. + expect(checkReportSupersession([{ reportId: "a" }, { reportId: "b" }], "a")).toEqual({ + _tag: "ok", + }); + }); + + it("rejects a fork and names the chain head", () => { + expect( + checkReportSupersession( + [ + { reportId: "a" }, + { reportId: "b", supersedesReportId: "a" }, + { reportId: "c", supersedesReportId: "b" }, + ], + "a", + ), + ).toEqual({ + _tag: "already-superseded", + supersededByReportId: "b", + chainHeadReportId: "c", + }); + }); + + it("reports an unknown report id", () => { + expect(checkReportSupersession([{ reportId: "a" }], "missing")).toEqual({ + _tag: "unknown-report", + }); + }); + + it("terminates on cyclic links instead of walking forever", () => { + // Unreachable through the decider's invariant, but this data comes back + // across a persistence boundary and a hang here would wedge command + // processing for the whole server. + const result = checkReportSupersession( + [ + { reportId: "a", supersedesReportId: "b" }, + { reportId: "b", supersedesReportId: "a" }, + ], + "a", + ); + expect(result._tag).toBe("already-superseded"); + }); +}); + +describe("reportAlreadySupersededMessage", () => { + it("names one report when the chain head is the direct superseder", () => { + const message = reportAlreadySupersededMessage({ + reportId: "a", + supersededByReportId: "b", + chainHeadReportId: "b", + }); + expect(message).toContain("a is already superseded by b"); + expect(message).toContain("supersede b instead"); + }); + + it("distinguishes the direct superseder from a further chain head", () => { + const message = reportAlreadySupersededMessage({ + reportId: "a", + supersededByReportId: "b", + chainHeadReportId: "c", + }); + expect(message).toContain("current head of that chain is c"); + expect(message).toContain("supersede c instead"); + }); +}); + describe("SpawnSessionInput", () => { it("decodes the git checkout specification", () => { expect( diff --git a/packages/contracts/src/sessionOrchestration.ts b/packages/contracts/src/sessionOrchestration.ts index 0639428ac257..238577023d40 100644 --- a/packages/contracts/src/sessionOrchestration.ts +++ b/packages/contracts/src/sessionOrchestration.ts @@ -361,6 +361,85 @@ export type ReadReportResult = typeof ReadReportResult.Type; export const supersededReportNotice = (supersededByReportId: string): string => `SUPERSEDED: this report was amended by a newer report on the same session. Read reportId "${supersededByReportId}" for the current account of the work; treat anything below as out of date.`; +/** + * Amendments form a single linear chain per thread — never a fork. + * + * Two reports both superseding A would leave "which report is current" + * ambiguous: reverse navigation from A could reach either, while latest-report + * selection (newest row) picks one of them, and the two answers need not + * agree. Rather than teach every reader a merge rule, superseding an + * already-superseded report is refused and the caller is pointed at the head + * of the chain. + * + * Structural on purpose: the decider checks this against folded read-model + * reports and the toolkit checks it against projection rows, and both must + * reach the same verdict from the same shape. + */ +export interface ReportSupersessionLink { + readonly reportId: string; + // Absent as `undefined` on event/read-model reports and as `null` on + // projection rows; both spellings mean "supersedes nothing", and neither + // can equal a report id, so the lookups below treat them identically. + readonly supersedesReportId?: string | null | undefined; +} + +export type ReportSupersessionCheck = + | { readonly _tag: "ok" } + // No report with that id on this thread. Deliberately does not distinguish + // "no such report anywhere" from "belongs to another thread": see + // SUPERSEDES_REPORT_NOT_FOUND_MESSAGE in the toolkit handlers. + | { readonly _tag: "unknown-report" } + | { + readonly _tag: "already-superseded"; + readonly supersededByReportId: string; + readonly chainHeadReportId: string; + }; + +export const checkReportSupersession = ( + reports: ReadonlyArray, + supersedesReportId: string, +): ReportSupersessionCheck => { + if (!reports.some((report) => report.reportId === supersedesReportId)) { + return { _tag: "unknown-report" }; + } + const supersederOf = (reportId: string) => + reports.find((report) => report.supersedesReportId === reportId); + const direct = supersederOf(supersedesReportId); + if (direct === undefined) { + return { _tag: "ok" }; + } + // Walk to the end of the chain so the caller is told where to actually + // attach, not merely that it lost. `seen` bounds the walk: the linear-chain + // invariant makes a cycle unreachable, but this data crosses a persistence + // boundary and an infinite loop in the decider would wedge command + // processing for the whole server. + let chainHead = direct; + const seen = new Set([supersedesReportId, direct.reportId]); + for (;;) { + const next = supersederOf(chainHead.reportId); + if (next === undefined || seen.has(next.reportId)) break; + seen.add(next.reportId); + chainHead = next; + } + return { + _tag: "already-superseded", + supersededByReportId: direct.reportId, + chainHeadReportId: chainHead.reportId, + }; +}; + +// One wording for the refusal, shared by the toolkit's friendly pre-check and +// the decider's authoritative rejection, so a caller cannot get two different +// explanations of the same state depending on which one caught it. +export const reportAlreadySupersededMessage = (input: { + readonly reportId: string; + readonly supersededByReportId: string; + readonly chainHeadReportId: string; +}): string => + input.chainHeadReportId === input.supersededByReportId + ? `Report ${input.reportId} is already superseded by ${input.supersededByReportId}; supersede ${input.supersededByReportId} instead. Amendments form a single linear chain, so only the newest report in a chain can be amended.` + : `Report ${input.reportId} is already superseded by ${input.supersededByReportId}, and the current head of that chain is ${input.chainHeadReportId}; supersede ${input.chainHeadReportId} instead. Amendments form a single linear chain, so only the newest report in a chain can be amended.`; + export const PingSessionInput = Schema.Struct({ threadId: ThreadId, }); @@ -513,6 +592,28 @@ export class SessionOrchestrationInvalidInputError extends Schema.TaggedErrorCla SessionOrchestrationErrorFields, ) {} +/** + * post_report refused to fork an amendment chain. + * + * Structured rather than prose alone because the caller's next move is + * mechanical: re-post against `chainHeadReportId`. It is also the losing side + * of a concurrent-amendment race, where the winner's id is the only thing the + * loser needs to make progress. + */ +export class SessionOrchestrationReportAlreadySupersededError extends Schema.TaggedErrorClass()( + "SessionOrchestrationReportAlreadySupersededError", + { + ...SessionOrchestrationErrorFields, + // The report the caller tried to supersede. + reportId: TrimmedNonEmptyString, + // The report that already superseded it. + supersededByReportId: TrimmedNonEmptyString, + // Newest report in that chain — what to supersede instead. Equal to + // supersededByReportId unless the chain has grown further. + chainHeadReportId: TrimmedNonEmptyString, + }, +) {} + export class SessionOrchestrationUnavailableError extends Schema.TaggedErrorClass()( "SessionOrchestrationUnavailableError", SessionOrchestrationErrorFields, @@ -529,5 +630,6 @@ export const SessionOrchestrationError = Schema.Union([ SessionOrchestrationUnavailableError, SessionOrchestrationOperationError, SessionOrchestrationWorktreeNotEmptyError, + SessionOrchestrationReportAlreadySupersededError, ]); export type SessionOrchestrationError = typeof SessionOrchestrationError.Type;