From 8861c66386fe5b4cf2a4b13bc719d3fa8df3b860 Mon Sep 17 00:00:00 2001 From: Jackson Miller Date: Sun, 30 Aug 2026 12:07:51 -0400 Subject: [PATCH] feat(j5): close archived participant obligations --- .../src/j5/a2a/ArchiveFactsService.test.ts | 230 ++++++++ apps/server/src/j5/a2a/ArchiveFactsService.ts | 184 ++++++ .../a2a/DeliveryTransport.integration.test.ts | 173 ++++++ apps/server/src/j5/a2a/DeliveryTransport.ts | 53 +- apps/server/src/j5/a2a/DeliveryWorker.test.ts | 155 ++++- apps/server/src/j5/a2a/DeliveryWorker.ts | 6 +- apps/server/src/j5/a2a/HumanInboxService.ts | 2 +- apps/server/src/j5/a2a/LedgerService.ts | 240 +++++--- .../src/j5/a2a/LifecycleService.test.ts | 553 ++++++++++++++++++ apps/server/src/j5/a2a/LifecycleService.ts | 466 +++++++++++++++ apps/server/src/j5/a2a/Migrations.test.ts | 267 ++++++++- apps/server/src/j5/a2a/Migrations.ts | 4 +- apps/server/src/j5/a2a/README.md | 10 + apps/server/src/j5/a2a/SendService.ts | 61 +- .../server/src/j5/a2a/SilenceDetector.test.ts | 147 ++++- apps/server/src/j5/a2a/SilenceDetector.ts | 80 +-- apps/server/src/j5/a2a/contracts.ts | 42 +- apps/server/src/j5/a2a/index.ts | 25 + .../j5/a2a/migrations/008_LifecycleClosure.ts | 329 +++++++++++ apps/server/src/j5/a2a/runtimeLayer.test.ts | 6 + apps/server/src/j5/a2a/runtimeLayer.ts | 13 + .../a2a-tickets/a3-silence-detector/index.md | 1 + .../a2a-tickets/a9-lifecycle-closure.md | 36 +- 23 files changed, 2921 insertions(+), 162 deletions(-) create mode 100644 apps/server/src/j5/a2a/ArchiveFactsService.test.ts create mode 100644 apps/server/src/j5/a2a/ArchiveFactsService.ts create mode 100644 apps/server/src/j5/a2a/LifecycleService.test.ts create mode 100644 apps/server/src/j5/a2a/LifecycleService.ts create mode 100644 apps/server/src/j5/a2a/migrations/008_LifecycleClosure.ts diff --git a/apps/server/src/j5/a2a/ArchiveFactsService.test.ts b/apps/server/src/j5/a2a/ArchiveFactsService.test.ts new file mode 100644 index 000000000000..03e88988965f --- /dev/null +++ b/apps/server/src/j5/a2a/ArchiveFactsService.test.ts @@ -0,0 +1,230 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as NodeSqliteClient from "../../persistence/NodeSqliteClient.ts"; +import { + A2AArchiveFacts, + A2AArchivePlacementFactsProviderError, + A2AArchivePlacementFactsProvider, + layer as archiveFactsLayer, + placementFactsUnavailableLayer, +} from "./ArchiveFactsService.ts"; +import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { runJ5A2AMigrations } from "./Migrations.ts"; +import { + CommCommandId, + ExchangeId, + ParticipantId, + SquadronId, + type AgentParticipant, +} from "./contracts.ts"; + +const timestamp = "2026-08-29T12:00:00.000Z"; +const squadronId = SquadronId.make("squadron:archive-facts"); +const participant: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:archive-facts:subject"), + threadId: ThreadId.make("thread:archive-facts:subject"), +}; +const outboundCounterparty: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:archive-facts:outbound"), + threadId: ThreadId.make("thread:archive-facts:outbound"), +}; +const inboundCounterparty: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:archive-facts:inbound"), + threadId: ThreadId.make("thread:archive-facts:inbound"), +}; + +const makeTestLayer = (placementLayer = placementFactsUnavailableLayer) => { + const database = NodeSqliteClient.layerMemory(); + const ledger = ledgerLayer.pipe(Layer.provide(database)); + const facts = archiveFactsLayer.pipe(Layer.provide(placementLayer), Layer.provide(database)); + return Layer.mergeAll(database, ledger, facts); +}; + +const seed = Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const ledger = yield* A2ALedger; + yield* ledger.createSquadron({ + squadron: { id: squadronId, name: "Archive facts", createdAt: timestamp }, + }); + for (const [index, candidate] of [ + participant, + outboundCounterparty, + inboundCounterparty, + ].entries()) { + yield* ledger.append({ + commandId: CommCommandId.make(`command:archive-facts:join:${index}`), + squadronId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: candidate.id, + exchangeId: null, + correlationId: null, + payload: { participant: candidate }, + createdAt: timestamp, + }, + }); + } + yield* ledger.append({ + commandId: CommCommandId.make("command:archive-facts:outbound"), + squadronId, + acceptedAt: timestamp, + event: { + kind: "exchange.opened", + sender: participant.id, + receiver: outboundCounterparty.id, + exchangeId: ExchangeId.make("exchange:archive-facts:outbound"), + correlationId: null, + payload: { intent: "Wait for a counterparty", urgency: "soon" }, + createdAt: "2026-08-29T12:00:01.000Z", + }, + }); + yield* ledger.append({ + commandId: CommCommandId.make("command:archive-facts:inbound"), + squadronId, + acceptedAt: timestamp, + event: { + kind: "exchange.opened", + sender: inboundCounterparty.id, + receiver: participant.id, + exchangeId: ExchangeId.make("exchange:archive-facts:inbound"), + correlationId: null, + payload: { intent: "Reply before retiring", urgency: "blocking" }, + createdAt: "2026-08-29T12:00:02.000Z", + }, + }); + yield* ledger.append({ + commandId: CommCommandId.make("command:archive-facts:unrelated"), + squadronId, + acceptedAt: timestamp, + event: { + kind: "exchange.opened", + sender: inboundCounterparty.id, + receiver: outboundCounterparty.id, + exchangeId: ExchangeId.make("exchange:archive-facts:unrelated"), + correlationId: null, + payload: { intent: "Unrelated work", urgency: null }, + createdAt: "2026-08-29T12:00:03.000Z", + }, + }); +}); + +it.effect( + "reports open exchanges in both directions and renders unavailable placement as unknown", + () => + Effect.gen(function* () { + yield* seed; + const sql = yield* SqlClient.SqlClient; + const before = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM j5_a2a_comm_event + `; + const facts = yield* (yield* A2AArchiveFacts).readForThread(participant.threadId); + const after = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM j5_a2a_comm_event + `; + + assert.equal(facts.state, "registered"); + if (facts.state !== "registered") return; + assert.deepStrictEqual( + facts.openExchanges.map((exchange) => ({ + exchangeId: exchange.exchangeId, + direction: exchange.direction, + replyObligation: exchange.replyObligation, + counterpartyId: exchange.counterpartyId, + })), + [ + { + exchangeId: "exchange:archive-facts:outbound", + direction: "outbound", + replyObligation: "counterparty-owes-reply", + counterpartyId: outboundCounterparty.id, + }, + { + exchangeId: "exchange:archive-facts:inbound", + direction: "inbound", + replyObligation: "participant-owes-reply", + counterpartyId: inboundCounterparty.id, + }, + ], + ); + assert.deepStrictEqual(facts.placementSubtree, { + state: "unknown", + reason: "placement-provider-unavailable", + }); + assert.deepStrictEqual(after, before); + }).pipe(Effect.provide(makeTestLayer())), +); + +it.effect("keeps a checked-empty placement subtree structurally distinct from unknown", () => + Effect.gen(function* () { + yield* seed; + const facts = yield* (yield* A2AArchiveFacts).readForThread(participant.threadId); + assert.equal(facts.state, "registered"); + if (facts.state !== "registered") return; + assert.deepStrictEqual(facts.placementSubtree, { state: "none" }); + }).pipe( + Effect.provide( + makeTestLayer( + Layer.succeed( + A2AArchivePlacementFactsProvider, + A2AArchivePlacementFactsProvider.of({ + readSubtree: () => Effect.succeed({ state: "none" }), + }), + ), + ), + ), + ), +); + +it.effect( + "renders a failed placement lookup as unknown instead of a reassuring empty subtree", + () => + Effect.gen(function* () { + yield* seed; + const facts = yield* (yield* A2AArchiveFacts).readForThread(participant.threadId); + assert.equal(facts.state, "registered"); + if (facts.state !== "registered") return; + assert.deepStrictEqual(facts.placementSubtree, { + state: "unknown", + reason: "placement-query-failed", + }); + }).pipe( + Effect.provide( + makeTestLayer( + Layer.succeed( + A2AArchivePlacementFactsProvider, + A2AArchivePlacementFactsProvider.of({ + readSubtree: () => + Effect.fail( + new A2AArchivePlacementFactsProviderError({ + operation: "read placement subtree", + cause: "placement store unavailable", + }), + ), + }), + ), + ), + ), + ), +); + +it.effect("names a native thread as not participating instead of guessing placement", () => + Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const threadId = ThreadId.make("thread:archive-facts:native"); + assert.deepStrictEqual(yield* (yield* A2AArchiveFacts).readForThread(threadId), { + state: "not-an-a2a-participant", + threadId, + openExchanges: [], + placementSubtree: { state: "not-applicable" }, + }); + }).pipe(Effect.provide(makeTestLayer())), +); diff --git a/apps/server/src/j5/a2a/ArchiveFactsService.ts b/apps/server/src/j5/a2a/ArchiveFactsService.ts new file mode 100644 index 000000000000..c5be72fad04c --- /dev/null +++ b/apps/server/src/j5/a2a/ArchiveFactsService.ts @@ -0,0 +1,184 @@ +import { type ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { resolveThreadHome } from "./HomeRegistrar.ts"; +import { ExchangeId, ParticipantId, SquadronId, type Urgency } from "./contracts.ts"; + +export type ArchivePlacementSubtree = + | { + readonly state: "unknown"; + readonly reason: "placement-provider-unavailable" | "placement-query-failed"; + } + | { readonly state: "none" } + | { + readonly state: "known"; + readonly participantIds: readonly [ParticipantId, ...Array]; + }; + +export class A2AArchivePlacementFactsProviderError extends Schema.TaggedErrorClass()( + "A2AArchivePlacementFactsProviderError", + { operation: Schema.String, cause: Schema.Defect() }, +) {} + +export interface A2AArchivePlacementFactsProviderShape { + readonly readSubtree: (input: { + readonly squadronId: SquadronId; + readonly participantId: ParticipantId; + }) => Effect.Effect; +} + +export class A2AArchivePlacementFactsProvider extends Context.Service< + A2AArchivePlacementFactsProvider, + A2AArchivePlacementFactsProviderShape +>()("t3/j5/a2a/ArchiveFactsService/A2AArchivePlacementFactsProvider") {} + +export const placementFactsUnavailableLayer = Layer.succeed( + A2AArchivePlacementFactsProvider, + A2AArchivePlacementFactsProvider.of({ + readSubtree: () => + Effect.succeed({ + state: "unknown", + reason: "placement-provider-unavailable", + }), + }), +); + +export interface OpenExchangeArchiveFact { + readonly squadronId: SquadronId; + readonly exchangeId: ExchangeId; + readonly direction: "inbound" | "outbound"; + readonly replyObligation: "participant-owes-reply" | "counterparty-owes-reply"; + readonly counterpartyId: ParticipantId; + readonly intent: string; + readonly urgency: Urgency | null; + readonly openedAt: string; +} + +export type ThreadPreArchiveFacts = + | { + readonly state: "not-an-a2a-participant"; + readonly threadId: ThreadId; + readonly openExchanges: readonly []; + readonly placementSubtree: { readonly state: "not-applicable" }; + } + | { + readonly state: "registered"; + readonly threadId: ThreadId; + readonly squadronId: SquadronId; + readonly participantId: ParticipantId; + readonly retired: boolean; + readonly openExchanges: ReadonlyArray; + readonly placementSubtree: ArchivePlacementSubtree; + }; + +export class A2AArchiveFactsError extends Schema.TaggedErrorClass()( + "A2AArchiveFactsError", + { operation: Schema.String, cause: Schema.Defect() }, +) {} + +export interface A2AArchiveFactsShape { + /** + * Read-only facts for the future human pre-archive dialog. This method does + * not decide whether archive or settle is allowed and performs no mutation. + */ + readonly readForThread: ( + threadId: ThreadId, + ) => Effect.Effect; +} + +export class A2AArchiveFacts extends Context.Service()( + "t3/j5/a2a/ArchiveFactsService/A2AArchiveFacts", +) {} + +interface ExchangeRow { + readonly squadron_id: string; + readonly exchange_id: string; + readonly sender_id: string; + readonly receiver_id: string; + readonly intent: string; + readonly urgency: Urgency | null; + readonly created_at: string; +} + +const archiveFactsError = (operation: string) => (cause: unknown) => + new A2AArchiveFactsError({ operation, cause }); + +export const layer = Layer.effect( + A2AArchiveFacts, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const placement = yield* A2AArchivePlacementFactsProvider; + + const readForThread: A2AArchiveFactsShape["readForThread"] = (threadId) => + Effect.gen(function* () { + const resolution = yield* resolveThreadHome(sql, threadId).pipe( + Effect.catchTag("A2AHomeNotFoundError", () => Effect.succeed(null)), + ); + if (resolution === null) { + return { + state: "not-an-a2a-participant", + threadId, + openExchanges: [], + placementSubtree: { state: "not-applicable" }, + } as const; + } + + const participantId = resolution.home.participantId; + const rows = yield* sql` + SELECT + squadron_id, + exchange_id, + sender_id, + receiver_id, + intent, + urgency, + created_at + FROM j5_a2a_exchange + WHERE status = 'open' + AND (sender_id = ${participantId} OR receiver_id = ${participantId}) + ORDER BY created_at, squadron_id, exchange_id + `; + const openExchanges = rows.map((row): OpenExchangeArchiveFact => { + const inbound = row.receiver_id === participantId; + return { + squadronId: SquadronId.make(row.squadron_id), + exchangeId: ExchangeId.make(row.exchange_id), + direction: inbound ? "inbound" : "outbound", + replyObligation: inbound ? "participant-owes-reply" : "counterparty-owes-reply", + counterpartyId: ParticipantId.make(inbound ? row.sender_id : row.receiver_id), + intent: row.intent, + urgency: row.urgency, + openedAt: row.created_at, + }; + }); + const placementSubtree = yield* placement + .readSubtree({ + squadronId: resolution.home.squadronId, + participantId, + }) + .pipe( + Effect.catchCause(() => + Effect.succeed({ + state: "unknown" as const, + reason: "placement-query-failed" as const, + }), + ), + ); + return { + state: "registered", + threadId, + squadronId: resolution.home.squadronId, + participantId, + retired: resolution.retired, + openExchanges, + placementSubtree, + } as const; + }).pipe(Effect.mapError(archiveFactsError("read pre-archive facts"))); + + return A2AArchiveFacts.of({ readForThread }); + }), +); diff --git a/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts b/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts index 0c9ff0badd28..2175455b5ff1 100644 --- a/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts +++ b/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts @@ -21,6 +21,7 @@ import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts"; import { ServerConfig } from "../../config.ts"; @@ -43,6 +44,10 @@ import { ThreadManagementService, type ThreadManagementSendMode, } from "../../orchestration-v2/ThreadManagementService.ts"; +import { + ThreadLifecycleService, + layer as threadLifecycleServiceLayer, +} from "../../orchestration-v2/ThreadLifecycleService.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import type { ProviderInstance } from "../../provider/ProviderDriver.ts"; import { ProviderInstanceRegistry } from "../../provider/Services/ProviderInstanceRegistry.ts"; @@ -54,14 +59,18 @@ import { deliveryMessageId, live as deliveryTransportLayer, } from "./DeliveryTransport.ts"; +import { manualLayer as deliveryWorkerLayer } from "./DeliveryWorker.ts"; import { formatHumanEnvelope, formatPeerEnvelope } from "./EnvelopeFormatter.ts"; import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { A2ALifecycleService, manualLayer as lifecycleServiceLayer } from "./LifecycleService.ts"; +import { A2ASenderRetiredError, A2ASendService, layer as sendServiceLayer } from "./SendService.ts"; import { CommCommandId, SquadronId, ExchangeId, LedgerMessageId, ParticipantId, + type AgentParticipant, } from "./contracts.ts"; const serverConfigLayer = ServerConfig.layerTest(process.cwd(), { @@ -245,6 +254,15 @@ const makeTestLayer = (harness: DeliveryHarness) => { ); }; +const makeLifecycleTestLayer = (harness: DeliveryHarness) => { + const base = makeTestLayer(harness); + const send = sendServiceLayer.pipe(Layer.provide(base)); + const worker = deliveryWorkerLayer.pipe(Layer.provide(base)); + const lifecycle = lifecycleServiceLayer.pipe(Layer.provide(worker), Layer.provide(base)); + const threadLifecycle = threadLifecycleServiceLayer.pipe(Layer.provide(base)); + return Layer.mergeAll(base, send, worker, lifecycle, threadLifecycle); +}; + const makeHarness = Effect.gen(function* () { return { deliveryInvocations: yield* Ref.make>([]), @@ -490,3 +508,158 @@ it.effect("attributes human-origin delivery to the user actor", () => }).pipe(Effect.provide(makeTestLayer(harness))); }), ); + +it.effect("routes real archive and delete commands through lifecycle closure exactly once", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const threads = yield* ThreadManagementService; + const threadLifecycle = yield* ThreadLifecycleService; + const ledger = yield* A2ALedger; + const send = yield* A2ASendService; + const lifecycle = yield* A2ALifecycleService; + const sql = yield* SqlClient.SqlClient; + const squadronId = SquadronId.make("squadron:j5-a2a-lifecycle-command-path"); + const sender: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:j5-a2a-lifecycle-command-path-sender"), + threadId: ThreadId.make("thread:j5-a2a-lifecycle-command-path-sender"), + }; + const receiver: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:j5-a2a-lifecycle-command-path-receiver"), + threadId: ThreadId.make("thread:j5-a2a-lifecycle-command-path-receiver"), + }; + const createdAt = "2026-08-29T14:00:00.000Z"; + + for (const [index, participant] of [sender, receiver].entries()) { + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:j5-a2a-lifecycle-command-path-create:${index}`), + threadId: participant.threadId, + projectId: ProjectId.make(`project:j5-a2a-lifecycle-command-path:${index}`), + title: `J5 A2A lifecycle command path ${index}`, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + } + yield* ledger.createSquadron({ + squadron: { id: squadronId, name: "Lifecycle command path", createdAt }, + }); + for (const [index, participant] of [sender, receiver].entries()) { + yield* ledger.append({ + commandId: CommCommandId.make(`command:j5-a2a-lifecycle-command-path-join:${index}`), + squadronId, + acceptedAt: createdAt, + event: { + kind: "participant.joined", + sender: null, + receiver: participant.id, + exchangeId: null, + correlationId: null, + payload: { participant }, + createdAt, + }, + }); + } + const opened = yield* send.send({ + commandId: CommCommandId.make("command:j5-a2a-lifecycle-command-path-open"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Archive the reply-owing participant through the real lifecycle command.", + expectReply: true, + intent: "Prove the production archive bridge", + acceptedAt: createdAt, + }); + + const storedLifecycleEvents = yield* threads.streamStoredEventsFrom().pipe( + Stream.filter( + (stored) => + stored.event.threadId === receiver.threadId && + (stored.event.type === "thread.archived" || stored.event.type === "thread.deleted"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }), + ); + yield* threadLifecycle.archive({ + commandId: CommandId.make("command:j5-a2a-lifecycle-command-path-archive"), + threadId: receiver.threadId, + }); + yield* threadLifecycle.unarchive({ + commandId: CommandId.make("command:j5-a2a-lifecycle-command-path-unarchive"), + threadId: receiver.threadId, + }); + yield* threadLifecycle.delete({ + commandId: CommandId.make("command:j5-a2a-lifecycle-command-path-delete"), + threadId: receiver.threadId, + }); + const stored = yield* Fiber.join(storedLifecycleEvents); + assert.deepStrictEqual( + Array.from(stored, (event) => event.event.type), + ["thread.archived", "thread.deleted"], + ); + for (const event of stored) yield* lifecycle.handleStoredEvent(event); + + const state = yield* sql<{ + readonly status: string; + readonly dropped_events: number; + readonly terminal_notices: number; + readonly participant_left_events: number; + }>` + SELECT + exchange.status, + ( + SELECT COUNT(*) + FROM j5_a2a_comm_event + WHERE kind = 'exchange.dropped' AND exchange_id = ${opened.exchangeId} + ) AS dropped_events, + ( + SELECT COUNT(*) + FROM j5_a2a_delivery + WHERE exchange_role = 'terminal_notice' AND exchange_id = ${opened.exchangeId} + ) AS terminal_notices, + ( + SELECT COUNT(*) + FROM j5_a2a_comm_event + WHERE kind = 'participant.left' AND receiver = ${receiver.id} + ) AS participant_left_events + FROM j5_a2a_exchange AS exchange + WHERE exchange.exchange_id = ${opened.exchangeId} + `; + assert.deepStrictEqual(state, [ + { + status: "dropped", + dropped_events: 1, + terminal_notices: 1, + participant_left_events: 1, + }, + ]); + assert.deepStrictEqual(yield* ledger.listMembership(squadronId), [ + { + squadronId, + participant: sender, + joinedSeq: 1, + updatedSeq: 1, + }, + ]); + const retiredSend = yield* Effect.flip( + send.send({ + commandId: CommCommandId.make("command:j5-a2a-lifecycle-command-path-retired-send"), + senderThreadId: receiver.threadId, + to: sender.id, + message: "Upstream unarchive must not revive A2A participation.", + acceptedAt: createdAt, + }), + ); + assert.instanceOf(retiredSend, A2ASenderRetiredError); + assert.include(retiredSend.message, "participant.left"); + }).pipe(Effect.provide(makeLifecycleTestLayer(harness))); + }), +); diff --git a/apps/server/src/j5/a2a/DeliveryTransport.ts b/apps/server/src/j5/a2a/DeliveryTransport.ts index 4d600c30c26d..0e4fde518520 100644 --- a/apps/server/src/j5/a2a/DeliveryTransport.ts +++ b/apps/server/src/j5/a2a/DeliveryTransport.ts @@ -86,7 +86,7 @@ interface MembershipRow { interface HumanExchangeRow { readonly sender_id: string; readonly receiver_id: string; - readonly status: "open" | "closed"; + readonly status: "open" | "closed" | "dropped"; readonly intent: string; readonly urgency: "blocking" | "soon" | "fyi" | null; readonly opened_seq: number; @@ -115,6 +115,7 @@ export const formatAgentDeliveryEnvelope = (input: AgentDeliveryInput): string = message: input.message, }); case "silence_notice": + case "lifecycle_notice": return input.message; default: return assertNever(input.envelopeChannel); @@ -171,7 +172,8 @@ export const live: Layer.Layer< attachments: [], mode, createdBy: - input.envelopeChannel === "silence_notice" + input.envelopeChannel === "silence_notice" || + input.envelopeChannel === "lifecycle_notice" ? "system" : isHumanParticipantId(input.senderId) ? "user" @@ -184,8 +186,10 @@ export const live: Layer.Layer< ), ), deliverHuman: (input) => - Effect.gen(function* () { - yield* sql` + input.envelopeChannel === "lifecycle_notice" + ? Effect.void + : Effect.gen(function* () { + yield* sql` INSERT INTO j5_a2a_human_inbox_data ( origin_squadron_id, message_id, @@ -205,8 +209,8 @@ export const live: Layer.Layer< ) ON CONFLICT(origin_squadron_id, message_id) DO NOTHING `; - if (input.exchangeId === null) return; - const exchanges = yield* sql` + if (input.exchangeId === null) return; + const exchanges = yield* sql` SELECT sender_id, receiver_id, status, intent, urgency, opened_seq, created_at FROM j5_a2a_exchange WHERE squadron_id = ${input.originSquadronId} @@ -214,18 +218,19 @@ export const live: Layer.Layer< AND receiver_id = ${input.receiverId} LIMIT 1 `; - const exchange = exchanges[0]; - if (exchange === undefined || exchange.urgency === null) { - return yield* new A2ADeliveryTargetError({ - participantId: input.receiverId, - state: "person-addressed exchange disappeared before inbox projection", - }); - } - // A follow-up may still be pending when the person answers. The raw - // message is durable above, but a discharged obligation has no active - // inbox row to update and is a successful no-op for projection. - if (exchange.status === "closed") return; - yield* sql` + const exchange = exchanges[0]; + if (exchange === undefined || exchange.urgency === null) { + return yield* new A2ADeliveryTargetError({ + participantId: input.receiverId, + state: "person-addressed exchange disappeared before inbox projection", + }); + } + // A follow-up may still be pending when the person answers or + // lifecycle closure drops the exchange. The raw peer message is + // durable above, but a discharged obligation has no active inbox + // row to update and is a successful projection no-op. + if (exchange.status !== "open") return; + yield* sql` INSERT INTO j5_a2a_human_inbox ( person_id, squadron_id, @@ -268,12 +273,12 @@ export const live: Layer.Layer< latest_message = excluded.latest_message WHERE j5_a2a_human_inbox.status = 'open' `; - }).pipe( - Effect.asVoid, - Effect.mapError( - (cause) => new A2ADeliveryTransportError({ operation: "deliver human", cause }), - ), - ), + }).pipe( + Effect.asVoid, + Effect.mapError( + (cause) => new A2ADeliveryTransportError({ operation: "deliver human", cause }), + ), + ), }); }), ); diff --git a/apps/server/src/j5/a2a/DeliveryWorker.test.ts b/apps/server/src/j5/a2a/DeliveryWorker.test.ts index f67aa28ecb5a..6f78baf77681 100644 --- a/apps/server/src/j5/a2a/DeliveryWorker.test.ts +++ b/apps/server/src/j5/a2a/DeliveryWorker.test.ts @@ -33,7 +33,15 @@ import { A2AHumanInbox, layer as humanInboxLayer } from "./HumanInboxService.ts" import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; import { runJ5A2AMigrations } from "./Migrations.ts"; import { A2ASendService, layer as sendLayer } from "./SendService.ts"; -import { CommCommandId, SquadronId, ParticipantId, type AgentParticipant } from "./contracts.ts"; +import { + CommCommandId, + CorrelationId, + LedgerMessageId, + LIFECYCLE_PARTICIPANT_ID, + SquadronId, + ParticipantId, + type AgentParticipant, +} from "./contracts.ts"; const timestamp = "2026-08-16T12:00:00.000Z"; const sender: AgentParticipant = { @@ -675,3 +683,148 @@ it.effect("treats a follow-up delivered after the human answered as a successful }).pipe(Effect.provide(layer)); }), ); + +it.effect("receipts a human lifecycle notice without creating a second actionable inbox row", () => + Effect.gen(function* () { + const database = NodeSqliteClient.layerMemory(); + const ledger = ledgerLayer.pipe(Layer.provide(database)); + const send = sendLayer.pipe(Layer.provide(ledger), Layer.provide(database)); + const transport = deliveryTransportLive.pipe( + Layer.provide(database), + Layer.provide(Layer.mock(ThreadManagementService)({})), + ); + const worker = deliveryWorkerLayerWithHooks(false).pipe( + Layer.provide(ledger), + Layer.provide(database), + Layer.provide(transport), + Layer.provide( + Layer.succeed( + A2ADeliveryHooks, + A2ADeliveryHooks.of({ afterTransportSuccess: () => Effect.void }), + ), + ), + ); + const layer = Layer.mergeAll(database, ledger, send, transport, worker); + + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const ledgerService = yield* A2ALedger; + const sendService = yield* A2ASendService; + const deliveryWorker = yield* A2ADeliveryWorker; + const sql = yield* SqlClient.SqlClient; + const squadronId = SquadronId.make("squadron:delivery:human-lifecycle"); + const noticeMessageId = LedgerMessageId.make("message:delivery:human-lifecycle"); + const correlationId = CorrelationId.make("correlation:delivery:human-lifecycle"); + yield* ledgerService.createSquadron({ + squadron: { id: squadronId, name: "Human lifecycle delivery", createdAt: timestamp }, + }); + yield* join(squadronId, sender, "human-lifecycle-sender"); + yield* sql` + INSERT INTO j5_a2a_human_person (person_id, is_local_operator, created_at) + VALUES (${person.id}, 1, ${timestamp}) + `; + const opened = yield* sendService.send({ + commandId: CommCommandId.make("command:delivery:human-lifecycle:open"), + senderThreadId: sender.threadId, + to: person.id, + message: "This person-addressed obligation will be dropped loudly.", + expectReply: true, + intent: "Prove lifecycle notices stay non-actionable", + urgency: "soon", + acceptedAt: timestamp, + }); + assert.equal((yield* deliveryWorker.runOnce)?.state, "delivered"); + yield* ledgerService.appendEvents({ + commandId: CommCommandId.make("command:delivery:human-lifecycle:drop"), + squadronId, + acceptedAt: timestamp, + events: [ + { + kind: "exchange.dropped", + sender: sender.id, + receiver: person.id, + exchangeId: opened.exchangeId!, + correlationId, + payload: { + disposition: "sender-retired", + cause: { + kind: "participant-archived", + participantId: sender.id, + squadronId, + }, + facts: { + replyRequired: false, + retryAllowed: false, + replacementRequired: false, + }, + noticeMessageId, + }, + createdAt: timestamp, + }, + { + kind: "message.sent", + sender: LIFECYCLE_PARTICIPANT_ID, + receiver: person.id, + exchangeId: opened.exchangeId!, + correlationId, + payload: { + messageId: noticeMessageId, + text: "The exchange was dropped because its agent sender retired from A2A.", + originSquadronId: squadronId, + receiverSquadronId: squadronId, + exchangeRole: "terminal_notice", + envelopeChannel: "lifecycle_notice", + }, + createdAt: timestamp, + }, + ], + }); + + const delivery = yield* deliveryWorker.runOnce; + assert.equal(delivery?.state, "delivered"); + const rawRows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_human_inbox_data + WHERE origin_squadron_id = ${squadronId} + `; + const inboxRows = yield* sql<{ + readonly cause_participant_id: string; + readonly replacement_required: number; + readonly reply_required: number; + readonly retry_allowed: number; + readonly status: string; + readonly terminal_disposition: string | null; + readonly terminal_notice_message_id: string | null; + }>` + SELECT + status, + terminal_disposition, + terminal_notice_message_id, + json_extract(terminal_cause, '$.participantId') AS cause_participant_id, + json_extract(terminal_facts, '$.replyRequired') AS reply_required, + json_extract(terminal_facts, '$.retryAllowed') AS retry_allowed, + json_extract(terminal_facts, '$.replacementRequired') AS replacement_required + FROM j5_a2a_human_inbox + WHERE person_id = ${person.id} AND exchange_id = ${opened.exchangeId!} + `; + const receipt = yield* sql<{ readonly status: string }>` + SELECT status + FROM j5_a2a_delivery + WHERE message_id = ${noticeMessageId} + `; + assert.deepStrictEqual(rawRows, [{ count: 1 }]); + assert.deepStrictEqual(inboxRows, [ + { + cause_participant_id: sender.id, + replacement_required: 0, + reply_required: 0, + retry_allowed: 0, + status: "dropped", + terminal_disposition: "sender-retired", + terminal_notice_message_id: noticeMessageId, + }, + ]); + assert.deepStrictEqual(receipt, [{ status: "delivered" }]); + }).pipe(Effect.provide(layer)); + }), +); diff --git a/apps/server/src/j5/a2a/DeliveryWorker.ts b/apps/server/src/j5/a2a/DeliveryWorker.ts index eb2361ae601f..5d2466f28068 100644 --- a/apps/server/src/j5/a2a/DeliveryWorker.ts +++ b/apps/server/src/j5/a2a/DeliveryWorker.ts @@ -39,8 +39,8 @@ interface DeliveryRow { readonly receiver_id: string; readonly receiver_squadron_id: string; readonly exchange_id: string | null; - readonly exchange_role: "none" | "ask" | "followup" | "reply"; - readonly envelope_channel: "peer" | "silence_notice"; + readonly exchange_role: "none" | "ask" | "followup" | "reply" | "terminal_notice"; + readonly envelope_channel: "peer" | "silence_notice" | "lifecycle_notice"; readonly correlation_id: string; readonly message_text: string; readonly status: "pending" | "retry_scheduled" | "delivered" | "alarmed"; @@ -161,7 +161,7 @@ const makeLayer = (daemon: boolean) => createdAt: receivedAt, }, ]; - if (exchangeId !== null) { + if (exchangeId !== null && row.exchange_role === "reply") { const exchanges = yield* sql` SELECT squadron_id, exchange_id, sender_id, receiver_id FROM j5_a2a_exchange diff --git a/apps/server/src/j5/a2a/HumanInboxService.ts b/apps/server/src/j5/a2a/HumanInboxService.ts index 73929e5fa683..c8ac67697150 100644 --- a/apps/server/src/j5/a2a/HumanInboxService.ts +++ b/apps/server/src/j5/a2a/HumanInboxService.ts @@ -65,7 +65,7 @@ interface ExchangeRow { readonly exchange_id: string; readonly sender_id: string; readonly receiver_id: string; - readonly status: "open" | "closed"; + readonly status: "open" | "closed" | "dropped"; } interface ExistingReplyRow { diff --git a/apps/server/src/j5/a2a/LedgerService.ts b/apps/server/src/j5/a2a/LedgerService.ts index e9fd2fc16e5a..c888e2d09b95 100644 --- a/apps/server/src/j5/a2a/LedgerService.ts +++ b/apps/server/src/j5/a2a/LedgerService.ts @@ -17,7 +17,9 @@ import { type CreateSquadronCommand, Squadron, ExchangeClosedPayload, + ExchangeDroppedPayload, ExchangeOpenedPayload, + type ExchangeId, MessageDeliveredPayload, MessageDeliveryFailedPayload, MessageSentPayload, @@ -103,6 +105,10 @@ export interface A2ALedgerShape { readonly appendEvents: ( command: AppendCommEventsCommand, ) => Effect.Effect; + readonly appendEventsIfExchangeOpen: ( + command: AppendCommEventsCommand, + exchangeId: ExchangeId, + ) => Effect.Effect; readonly readEvents: (input: { readonly squadronId: SquadronId; readonly cursor: LedgerCursor; @@ -165,6 +171,7 @@ const decodeReceipt = Schema.decodeUnknownEffect(CommCommandReceipt); const decodeMembership = Schema.decodeUnknownEffect(Membership); const decodeExchangeOpened = Schema.decodeUnknownEffect(ExchangeOpenedPayload); const decodeExchangeClosed = Schema.decodeUnknownEffect(ExchangeClosedPayload); +const decodeExchangeDropped = Schema.decodeUnknownEffect(ExchangeDroppedPayload); const decodeMessageSent = Schema.decodeUnknownEffect(MessageSentPayload); const decodeMessageDelivered = Schema.decodeUnknownEffect(MessageDeliveredPayload); const decodeMessageDeliveryFailed = Schema.decodeUnknownEffect(MessageDeliveryFailedPayload); @@ -177,7 +184,11 @@ const preserveDomainError = isA2ALedgerError(cause) ? cause : new A2AStorageError({ operation, cause }); const squadronFromRow = (row: SquadronRow) => - decodeSquadron({ id: row.id, name: row.name, createdAt: row.created_at }); + decodeSquadron({ + id: row.id, + name: row.name, + createdAt: row.created_at, + }); const eventFromRow = Effect.fn("j5.a2a.eventFromRow")(function* (row: EventRow) { return yield* decodeStoredEvent({ @@ -343,6 +354,45 @@ export const layer: Layer.Layer = Layer.e `; return; } + case "exchange.dropped": { + const dropped = yield* decodeExchangeDropped(event.payload); + if (event.exchangeId === null) { + return yield* new A2AStorageError({ operation: "project dropped exchange" }); + } + const rows = yield* sql<{ readonly exchange_id: string }>` + UPDATE j5_a2a_exchange + SET + status = 'dropped', + closed_seq = ${event.seq}, + updated_at = ${event.createdAt} + WHERE squadron_id = ${event.squadronId} + AND exchange_id = ${event.exchangeId} + AND status = 'open' + RETURNING exchange_id + `; + if (rows[0] === undefined) { + return yield* new A2AStorageError({ operation: "project dropped exchange" }); + } + const terminalCause = yield* encodeJson(dropped.cause); + const terminalFacts = yield* encodeJson(dropped.facts); + // A4 owns this retained projection; the ledger applies its terminal + // state from the ordinary exchange.dropped fact in the same commit. + yield* sql` + UPDATE j5_a2a_human_inbox + SET + status = 'dropped', + terminal_seq = ${event.seq}, + terminal_at = ${event.createdAt}, + terminal_disposition = ${dropped.disposition}, + terminal_cause = ${terminalCause}, + terminal_facts = ${terminalFacts}, + terminal_notice_message_id = ${dropped.noticeMessageId} + WHERE squadron_id = ${event.squadronId} + AND exchange_id = ${event.exchangeId} + AND status = 'open' + `; + return; + } case "message.sent": { const payload = yield* decodeMessageSent(event.payload); if (event.sender === null || event.receiver === null || event.correlationId === null) { @@ -467,23 +517,21 @@ export const layer: Layer.Layer = Layer.e }, ); - const appendEventsEffect = Effect.fn("j5.a2a.appendEvents")(function* ( + const appendEventsInTransaction = Effect.fn("j5.a2a.appendEventsInTransaction")(function* ( command: AppendCommEventsCommand, ) { - const result = yield* sql.withTransaction( - Effect.gen(function* () { - yield* ensureSquadron(command.squadronId); - const sequenceRows = yield* sql<{ readonly next_seq: number }>` + yield* ensureSquadron(command.squadronId); + const sequenceRows = yield* sql<{ readonly next_seq: number }>` SELECT COALESCE(MAX(seq), 0) + 1 AS next_seq FROM j5_a2a_comm_event WHERE squadron_id = ${command.squadronId} `; - const firstSeq = sequenceRows[0]?.next_seq; - if (firstSeq === undefined) { - return yield* new A2AStorageError({ operation: "allocate communication sequences" }); - } - const resultSeq = firstSeq + command.events.length - 1; - const reserved = yield* sql<{ readonly command_id: string }>` + const firstSeq = sequenceRows[0]?.next_seq; + if (firstSeq === undefined) { + return yield* new A2AStorageError({ operation: "allocate communication sequences" }); + } + const resultSeq = firstSeq + command.events.length - 1; + const reserved = yield* sql<{ readonly command_id: string }>` INSERT INTO j5_a2a_comm_command_receipt ( command_id, squadron_id, @@ -501,25 +549,25 @@ export const layer: Layer.Layer = Layer.e RETURNING command_id `; - if (reserved[0] === undefined) { - const receiptRows = yield* sql` + if (reserved[0] === undefined) { + const receiptRows = yield* sql` SELECT command_id, squadron_id, command_type, accepted_at, result_seq FROM j5_a2a_comm_command_receipt WHERE command_id = ${command.commandId} LIMIT 1 `; - const row = receiptRows[0]; - if (row === undefined) { - return yield* new A2AStorageError({ operation: "read replayed batch receipt" }); - } - if (row.squadron_id !== command.squadronId) { - return yield* new CommCommandConflictError({ - commandId: command.commandId, - requestedSquadronId: command.squadronId, - existingSquadronId: row.squadron_id, - }); - } - const eventRows = yield* sql` + const row = receiptRows[0]; + if (row === undefined) { + return yield* new A2AStorageError({ operation: "read replayed batch receipt" }); + } + if (row.squadron_id !== command.squadronId) { + return yield* new CommCommandConflictError({ + commandId: command.commandId, + requestedSquadronId: command.squadronId, + existingSquadronId: row.squadron_id, + }); + } + const eventRows = yield* sql` SELECT seq, squadron_id, @@ -534,35 +582,35 @@ export const layer: Layer.Layer = Layer.e WHERE squadron_id = ${command.squadronId} AND command_id = ${command.commandId} ORDER BY seq `; - if (eventRows.length === 0) { - return yield* new A2AStorageError({ operation: "read replayed batch events" }); - } - return { - receipt: yield* receiptFromRow(row), - events: yield* Effect.forEach(eventRows, eventFromRow, { concurrency: 1 }), - committed: false as const, - }; - } + if (eventRows.length === 0) { + return yield* new A2AStorageError({ operation: "read replayed batch events" }); + } + return { + receipt: yield* receiptFromRow(row), + events: yield* Effect.forEach(eventRows, eventFromRow, { concurrency: 1 }), + committed: false as const, + }; + } - const events: Array = []; - for (const [index, candidate] of command.events.entries()) { - if ( - (candidate.kind === "participant.joined" || candidate.kind === "participant.left") && - candidate.payload.participant.kind === "human" - ) { - return yield* new A2AStorageError({ - operation: "append host-global human as Squadron membership", - }); - } - const pending = decideAppendCommEvent({ - commandId: command.commandId, - squadronId: command.squadronId, - acceptedAt: command.acceptedAt, - event: candidate, - })[0]; - const seq = firstSeq + index; - const payload = yield* encodeJson(pending.payload); - yield* sql` + const events: Array = []; + for (const [index, candidate] of command.events.entries()) { + if ( + (candidate.kind === "participant.joined" || candidate.kind === "participant.left") && + candidate.payload.participant.kind === "human" + ) { + return yield* new A2AStorageError({ + operation: "append host-global human as Squadron membership", + }); + } + const pending = decideAppendCommEvent({ + commandId: command.commandId, + squadronId: command.squadronId, + acceptedAt: command.acceptedAt, + event: candidate, + })[0]; + const seq = firstSeq + index; + const payload = yield* encodeJson(pending.payload); + yield* sql` INSERT INTO j5_a2a_comm_event ( seq, squadron_id, @@ -587,24 +635,27 @@ export const layer: Layer.Layer = Layer.e ${command.commandId} ) `; - const event = yield* decodeStoredEvent({ seq, ...pending }); - yield* applyMembership(event); - yield* applyA2Projection(event, command.commandId); - events.push(event); - } - return { - receipt: yield* decodeReceipt({ - commandId: command.commandId, - squadronId: command.squadronId, - commandType: "comm.append", - acceptedAt: command.acceptedAt, - resultSeq, - }), - events, - committed: true as const, - }; + const event = yield* decodeStoredEvent({ seq, ...pending }); + yield* applyMembership(event); + yield* applyA2Projection(event, command.commandId); + events.push(event); + } + return { + receipt: yield* decodeReceipt({ + commandId: command.commandId, + squadronId: command.squadronId, + commandType: "comm.append", + acceptedAt: command.acceptedAt, + resultSeq, }), - ); + events, + committed: true as const, + }; + }); + + const publishCommitted = Effect.fn("j5.a2a.publishCommitted")(function* ( + result: AppendEventsResult, + ) { if (result.committed) { for (const event of result.events) { yield* PubSub.publish(committed, event); @@ -613,12 +664,44 @@ export const layer: Layer.Layer = Layer.e return result; }); + const appendEventsEffect = Effect.fn("j5.a2a.appendEvents")(function* ( + command: AppendCommEventsCommand, + ) { + return yield* sql + .withTransaction(appendEventsInTransaction(command)) + .pipe(Effect.flatMap(publishCommitted)); + }); + + const appendEventsIfExchangeOpenEffect = Effect.fn("j5.a2a.appendEventsIfExchangeOpen")( + function* (command: AppendCommEventsCommand, exchangeId: ExchangeId) { + const result = yield* sql.withTransaction( + Effect.gen(function* () { + const open = yield* sql<{ readonly exchange_id: string }>` + SELECT exchange_id + FROM j5_a2a_exchange + WHERE squadron_id = ${command.squadronId} + AND exchange_id = ${exchangeId} + AND status = 'open' + LIMIT 1 + `; + if (open[0] === undefined) return null; + return yield* appendEventsInTransaction(command); + }), + ); + return result === null ? null : yield* publishCommitted(result); + }, + ); + return A2ALedger.of({ createSquadron: (command) => Effect.gen(function* () { yield* sql` INSERT INTO j5_a2a_squadron (id, name, created_at) - VALUES (${command.squadron.id}, ${command.squadron.name}, ${command.squadron.createdAt}) + VALUES ( + ${command.squadron.id}, + ${command.squadron.name}, + ${command.squadron.createdAt} + ) ON CONFLICT(id) DO NOTHING `; return yield* squadronFromRow( @@ -633,14 +716,19 @@ export const layer: Layer.Layer = Layer.e listSquadrons: () => Effect.gen(function* () { const rows = yield* sql` - SELECT id, name, created_at FROM j5_a2a_squadron ORDER BY created_at, id + SELECT id, name, created_at + FROM j5_a2a_squadron + ORDER BY created_at, id `; return yield* Effect.forEach(rows, squadronFromRow, { concurrency: 1 }); }).pipe(Effect.mapError(preserveDomainError("list squadrons"))), readSquadron: (squadronId) => Effect.gen(function* () { const rows = yield* sql` - SELECT id, name, created_at FROM j5_a2a_squadron WHERE id = ${squadronId} LIMIT 1 + SELECT id, name, created_at + FROM j5_a2a_squadron + WHERE id = ${squadronId} + LIMIT 1 `; const row = rows[0]; if (row === undefined) return yield* new SquadronNotFoundError({ squadronId }); @@ -672,6 +760,10 @@ export const layer: Layer.Layer = Layer.e appendPermit .withPermit(appendEventsEffect(command)) .pipe(Effect.mapError(preserveDomainError("append communication events"))), + appendEventsIfExchangeOpen: (command, exchangeId) => + appendPermit + .withPermit(appendEventsIfExchangeOpenEffect(command, exchangeId)) + .pipe(Effect.mapError(preserveDomainError("append communication events if open"))), readEvents: ({ squadronId, cursor, limit }) => Effect.gen(function* () { yield* ensureSquadron(squadronId); diff --git a/apps/server/src/j5/a2a/LifecycleService.test.ts b/apps/server/src/j5/a2a/LifecycleService.test.ts new file mode 100644 index 000000000000..a7c72e2011f6 --- /dev/null +++ b/apps/server/src/j5/a2a/LifecycleService.test.ts @@ -0,0 +1,553 @@ +import { assert, it } from "@effect/vitest"; +import { EventId, type OrchestrationV2StoredEvent, ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as NodeSqliteClient from "../../persistence/NodeSqliteClient.ts"; +import { ThreadManagementService } from "../../orchestration-v2/ThreadManagementService.ts"; +import { + A2ADeliveryTransport, + type AgentDeliveryInput, + type HumanDeliveryInput, +} from "./DeliveryTransport.ts"; +import { A2ADeliveryWorker, manualLayer as deliveryWorkerLayer } from "./DeliveryWorker.ts"; +import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { A2ALifecycleService, manualLayer as lifecycleLayer } from "./LifecycleService.ts"; +import { runJ5A2AMigrations } from "./Migrations.ts"; +import { + A2AParticipantArchivedError, + A2ASenderRetiredError, + A2ASendService, + layer as sendLayer, +} from "./SendService.ts"; +import { + CommCommandId, + ExchangeDroppedPayload, + SquadronId, + ParticipantId, + type AgentParticipant, + type HumanParticipant, +} from "./contracts.ts"; + +const openedAt = "2026-08-23T12:00:00.000Z"; +const archivedAt = "2026-08-23T12:05:00.000Z"; +const decodeExchangeDroppedPayload = Schema.decodeUnknownEffect( + Schema.fromJsonString(ExchangeDroppedPayload), +); + +const sender: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:lifecycle:sender"), + threadId: ThreadId.make("thread:lifecycle:sender"), +}; +const receiver: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:lifecycle:receiver"), + threadId: ThreadId.make("thread:lifecycle:receiver"), +}; +const human: HumanParticipant = { + kind: "human", + id: ParticipantId.make("human:lifecycle:person"), +}; + +interface DeliveredNotice { + readonly channel: "agent" | "human"; + readonly input: AgentDeliveryInput | HumanDeliveryInput; +} + +const makeTestLayer = ( + notices: Ref.Ref>, + storedEvents: Stream.Stream = Stream.never, +) => { + const database = NodeSqliteClient.layerMemory(); + const ledger = ledgerLayer.pipe(Layer.provide(database)); + const send = sendLayer.pipe(Layer.provide(ledger), Layer.provide(database)); + const transport = Layer.succeed( + A2ADeliveryTransport, + A2ADeliveryTransport.of({ + deliverAgent: (input) => + Ref.update(notices, (current) => [...current, { channel: "agent" as const, input }]), + deliverHuman: (input) => + Ref.update(notices, (current) => [...current, { channel: "human" as const, input }]), + }), + ); + const worker = deliveryWorkerLayer.pipe( + Layer.provide(ledger), + Layer.provide(database), + Layer.provide(transport), + ); + const threadManagement = Layer.mock(ThreadManagementService)({ + streamStoredEventsFrom: ({ afterSequence } = {}) => + storedEvents.pipe(Stream.filter((event) => event.sequence > (afterSequence ?? 0))), + }); + const lifecycle = lifecycleLayer.pipe( + Layer.provide(ledger), + Layer.provide(worker), + Layer.provide(database), + Layer.provide(threadManagement), + ); + return Layer.mergeAll(database, ledger, send, transport, worker, threadManagement, lifecycle); +}; + +const retiredThreadEvent = ( + type: "thread.archived" | "thread.deleted", + threadId: ThreadId, + sequence: number, +): OrchestrationV2StoredEvent => + ({ + sequence, + commandId: null, + event: { + id: EventId.make(`event:lifecycle:${type}:${sequence}`), + type, + threadId, + occurredAt: DateTime.makeUnsafe(archivedAt), + payload: { id: threadId }, + }, + }) as unknown as OrchestrationV2StoredEvent; + +const createSquadron = Effect.fn("test.j5.a2a.lifecycle.createSquadron")(function* ( + squadronId: SquadronId, + name: string, +) { + yield* (yield* A2ALedger).createSquadron({ + squadron: { id: squadronId, name, createdAt: openedAt }, + }); +}); + +const join = Effect.fn("test.j5.a2a.lifecycle.join")(function* ( + squadronId: SquadronId, + participant: AgentParticipant, + suffix: string, +) { + yield* (yield* A2ALedger).append({ + commandId: CommCommandId.make(`command:lifecycle:join:${suffix}`), + squadronId, + acceptedAt: openedAt, + event: { + kind: "participant.joined", + sender: null, + receiver: participant.id, + exchangeId: null, + correlationId: null, + payload: { participant }, + createdAt: openedAt, + }, + }); +}); + +it.effect("drops a receiver-retired exchange loudly and rejects new sends to the archive", () => + Effect.gen(function* () { + const notices = yield* Ref.make>([]); + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const squadronId = SquadronId.make("squadron:lifecycle:receiver-retired"); + yield* createSquadron(squadronId, "Receiver retired"); + yield* join(squadronId, sender, "receiver-retired:sender"); + yield* join(squadronId, receiver, "receiver-retired:receiver"); + const send = yield* A2ASendService; + const lifecycle = yield* A2ALifecycleService; + const sql = yield* SqlClient.SqlClient; + const opened = yield* send.send({ + commandId: CommCommandId.make("command:lifecycle:receiver-retired:open"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Please answer before retirement.", + expectReply: true, + intent: "Prove receiver retirement closure", + acceptedAt: openedAt, + }); + assert.isNotNull(opened.exchangeId); + const exchangeId = opened.exchangeId!; + + const first = yield* lifecycle.archiveParticipant({ + participantId: receiver.id, + archivedAt, + }); + const replay = yield* lifecycle.archiveParticipant({ + participantId: receiver.id, + archivedAt: "2026-08-23T12:06:00.000Z", + }); + assert.deepStrictEqual(first.droppedExchangeIds, [exchangeId]); + assert.isTrue(first.archived); + assert.deepStrictEqual(replay, { archived: false, droppedExchangeIds: [] }); + + const exchange = yield* sql<{ + readonly status: string; + readonly closed_seq: number | null; + }>` + SELECT status, closed_seq + FROM j5_a2a_exchange + WHERE exchange_id = ${exchangeId} + `; + assert.equal(exchange[0]?.status, "dropped"); + assert.isNumber(exchange[0]?.closed_seq); + + const events = yield* sql<{ + readonly kind: string; + readonly sender: string | null; + readonly receiver: string | null; + readonly payload: string; + }>` + SELECT kind, sender, receiver, payload + FROM j5_a2a_comm_event + WHERE exchange_id = ${exchangeId} + AND kind IN ('exchange.dropped', 'message.sent') + ORDER BY seq + `; + const dropped = events.find((event) => event.kind === "exchange.dropped"); + assert.equal(dropped?.sender, sender.id); + assert.equal(dropped?.receiver, receiver.id); + const droppedPayload = yield* decodeExchangeDroppedPayload(dropped!.payload); + assert.deepInclude(droppedPayload, { + disposition: "receiver-retired", + facts: { + replyRequired: false, + retryAllowed: false, + replacementRequired: false, + }, + }); + assert.lengthOf( + events.filter((event) => event.kind === "exchange.dropped"), + 1, + ); + const retirement = yield* sql<{ + readonly seq: number; + readonly kind: string; + readonly squadron_id: string; + readonly participant_id: string; + readonly participant_kind: string; + readonly thread_id: string; + }>` + SELECT + seq, + kind, + squadron_id, + json_extract(payload, '$.participant.id') AS participant_id, + json_extract(payload, '$.participant.kind') AS participant_kind, + json_extract(payload, '$.participant.threadId') AS thread_id + FROM j5_a2a_comm_event + WHERE kind = 'participant.left' AND receiver = ${receiver.id} + `; + const joinedSequence = yield* sql<{ readonly seq: number }>` + SELECT seq + FROM j5_a2a_comm_event + WHERE kind = 'participant.joined' AND receiver = ${receiver.id} + `; + assert.lengthOf(retirement, 1); + assert.deepInclude(retirement[0]!, { + kind: "participant.left", + squadron_id: squadronId, + participant_id: receiver.id, + participant_kind: "agent", + thread_id: receiver.threadId, + }); + assert.isAbove(retirement[0]!.seq, joinedSequence[0]!.seq); + + const error = yield* Effect.flip( + send.send({ + commandId: CommCommandId.make("command:lifecycle:receiver-retired:after"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "This must be rejected.", + acceptedAt: archivedAt, + }), + ); + assert.instanceOf(error, A2AParticipantArchivedError); + assert.include(error.message, "A2A-retired"); + + const milestones = yield* (yield* A2ADeliveryWorker).drain; + assert.isTrue(milestones.some((milestone) => milestone.state === "delivered")); + const delivered = yield* Ref.get(notices); + assert.isTrue( + delivered.some( + (notice) => + notice.input.envelopeChannel === "lifecycle_notice" && + notice.input.receiverId === sender.id, + ), + ); + }).pipe(Effect.provide(makeTestLayer(notices))); + }), +); + +it.effect("drops a sender-retired person exchange so the active inbox source is terminal", () => + Effect.gen(function* () { + const notices = yield* Ref.make>([]); + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const squadronId = SquadronId.make("squadron:lifecycle:person"); + yield* createSquadron(squadronId, "Person inbox closure"); + yield* join(squadronId, sender, "person:sender"); + yield* (yield* SqlClient.SqlClient)` + INSERT INTO j5_a2a_human_person (person_id, is_local_operator, created_at) + VALUES (${human.id}, 1, ${openedAt}) + `; + const opened = yield* (yield* A2ASendService).send({ + commandId: CommCommandId.make("command:lifecycle:person:open"), + senderThreadId: sender.threadId, + to: human.id, + message: "A person should not owe an archived asker.", + expectReply: true, + intent: "Prove person inbox closure", + urgency: "blocking", + acceptedAt: openedAt, + }); + + yield* (yield* A2ALifecycleService).archiveParticipant({ + participantId: sender.id, + archivedAt, + }); + const retiredError = yield* Effect.flip( + (yield* A2ASendService).send({ + commandId: CommCommandId.make("command:lifecycle:person:retired-send"), + senderThreadId: sender.threadId, + to: human.id, + message: "Retired send must fail.", + acceptedAt: archivedAt, + }), + ); + assert.instanceOf(retiredError, A2ASenderRetiredError); + assert.include(retiredError.message, "participant.left"); + const source = yield* (yield* SqlClient.SqlClient)<{ + readonly status: string; + readonly disposition: string; + }>` + SELECT + exchange.status, + json_extract(event.payload, '$.disposition') AS disposition + FROM j5_a2a_exchange AS exchange + JOIN j5_a2a_comm_event AS event + ON event.squadron_id = exchange.squadron_id + AND event.exchange_id = exchange.exchange_id + AND event.kind = 'exchange.dropped' + WHERE exchange.exchange_id = ${opened.exchangeId} + `; + assert.deepStrictEqual(source, [{ status: "dropped", disposition: "sender-retired" }]); + yield* (yield* A2ADeliveryWorker).drain; + const delivered = yield* Ref.get(notices); + assert.isTrue( + delivered.some( + (notice) => + notice.channel === "human" && + notice.input.envelopeChannel === "lifecycle_notice" && + notice.input.receiverId === human.id, + ), + ); + }).pipe(Effect.provide(makeTestLayer(notices))); + }), +); + +it.effect("leaves unrelated same- and cross-Squadron exchanges open", () => + Effect.gen(function* () { + const notices = yield* Ref.make>([]); + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const localSquadronId = SquadronId.make("squadron:lifecycle:healthy:local"); + const remoteSquadronId = SquadronId.make("squadron:lifecycle:healthy:remote"); + const localPeer: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:lifecycle:healthy:local"), + threadId: ThreadId.make("thread:lifecycle:healthy:local"), + }; + const secondLocalPeer: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:lifecycle:healthy:second-local"), + threadId: ThreadId.make("thread:lifecycle:healthy:second-local"), + }; + const remotePeer: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:lifecycle:healthy:remote"), + threadId: ThreadId.make("thread:lifecycle:healthy:remote"), + }; + yield* createSquadron(localSquadronId, "Healthy local"); + yield* createSquadron(remoteSquadronId, "Healthy remote"); + yield* join(localSquadronId, sender, "healthy:affected-sender"); + yield* join(localSquadronId, receiver, "healthy:affected-receiver"); + yield* join(localSquadronId, localPeer, "healthy:local"); + yield* join(localSquadronId, secondLocalPeer, "healthy:second-local"); + yield* join(remoteSquadronId, remotePeer, "healthy:remote"); + const send = yield* A2ASendService; + const affected = yield* send.send({ + commandId: CommCommandId.make("command:lifecycle:healthy:affected"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "This obligation should drop.", + expectReply: true, + intent: "Affected exchange", + acceptedAt: openedAt, + }); + const sameSquadron = yield* send.send({ + commandId: CommCommandId.make("command:lifecycle:healthy:same"), + senderThreadId: localPeer.threadId, + to: secondLocalPeer.id, + message: "This same-Squadron exchange stays open.", + expectReply: true, + intent: "Healthy same-Squadron exchange", + acceptedAt: openedAt, + }); + const crossSquadron = yield* send.send({ + commandId: CommCommandId.make("command:lifecycle:healthy:cross"), + senderThreadId: localPeer.threadId, + to: remotePeer.id, + message: "This cross-Squadron exchange stays open.", + expectReply: true, + intent: "Healthy cross-Squadron exchange", + acceptedAt: openedAt, + }); + + yield* (yield* A2ALifecycleService).archiveParticipant({ + participantId: receiver.id, + archivedAt, + }); + + const rows = yield* (yield* SqlClient.SqlClient)<{ + readonly exchange_id: string; + readonly status: string; + }>` + SELECT exchange_id, status + FROM j5_a2a_exchange + WHERE exchange_id IN (${affected.exchangeId}, ${sameSquadron.exchangeId}, ${crossSquadron.exchangeId}) + ORDER BY exchange_id + `; + assert.deepStrictEqual(Object.fromEntries(rows.map((row) => [row.exchange_id, row.status])), { + [affected.exchangeId!]: "dropped", + [sameSquadron.exchangeId!]: "open", + [crossSquadron.exchangeId!]: "open", + }); + }).pipe(Effect.provide(makeTestLayer(notices))); + }), +); + +it.effect("replays archive then delete from its cursor without duplicating closure", () => + Effect.gen(function* () { + const notices = yield* Ref.make>([]); + const nativeNoHome = retiredThreadEvent( + "thread.archived", + ThreadId.make("thread:lifecycle:no-home"), + 6, + ); + const archived = retiredThreadEvent("thread.archived", receiver.threadId, 7); + const deleted = retiredThreadEvent("thread.deleted", receiver.threadId, 8); + const storedEvents = Stream.fromIterable([nativeNoHome, archived, deleted]); + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const squadronId = SquadronId.make("squadron:lifecycle:bridge"); + yield* createSquadron(squadronId, "Archive bridge"); + yield* join(squadronId, sender, "bridge:sender"); + yield* join(squadronId, receiver, "bridge:receiver"); + const opened = yield* (yield* A2ASendService).send({ + commandId: CommCommandId.make("command:lifecycle:bridge:open"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "The committed archive event should close this.", + expectReply: true, + intent: "Prove archive bridge replay", + acceptedAt: openedAt, + }); + assert.isNotNull(opened.exchangeId); + + const lifecycle = yield* A2ALifecycleService; + yield* lifecycle.replayCommittedEvents; + yield* lifecycle.replayCommittedEvents; + + const state = yield* (yield* SqlClient.SqlClient)<{ + readonly status: string; + readonly dropped_events: number; + readonly terminal_notices: number; + readonly after_sequence: number; + }>` + SELECT + exchange.status, + ( + SELECT COUNT(*) + FROM j5_a2a_comm_event + WHERE kind = 'exchange.dropped' AND exchange_id = ${opened.exchangeId} + ) AS dropped_events, + ( + SELECT COUNT(*) + FROM j5_a2a_delivery + WHERE exchange_role = 'terminal_notice' AND exchange_id = ${opened.exchangeId} + ) AS terminal_notices, + cursor.after_sequence + FROM j5_a2a_exchange AS exchange + CROSS JOIN j5_a2a_lifecycle_cursor AS cursor + WHERE exchange.exchange_id = ${opened.exchangeId} + `; + assert.deepStrictEqual(state, [ + { + status: "dropped", + dropped_events: 1, + terminal_notices: 1, + after_sequence: 8, + }, + ]); + }).pipe(Effect.provide(makeTestLayer(notices, storedEvents))); + }), +); + +it.effect("drops a reply-owing participant exactly once from committed thread deletion", () => + Effect.gen(function* () { + const notices = yield* Ref.make>([]); + const storedEvents = Stream.make(retiredThreadEvent("thread.deleted", receiver.threadId, 11)); + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const squadronId = SquadronId.make("squadron:lifecycle:delete-bridge"); + yield* createSquadron(squadronId, "Delete bridge"); + yield* join(squadronId, sender, "delete-bridge:sender"); + yield* join(squadronId, receiver, "delete-bridge:receiver"); + const opened = yield* (yield* A2ASendService).send({ + commandId: CommCommandId.make("command:lifecycle:delete-bridge:open"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Deleting the receiver must end its reply obligation.", + expectReply: true, + intent: "Prove deletion bridge closure", + acceptedAt: openedAt, + }); + assert.isNotNull(opened.exchangeId); + + const lifecycle = yield* A2ALifecycleService; + yield* lifecycle.replayCommittedEvents; + yield* lifecycle.replayCommittedEvents; + + const state = yield* (yield* SqlClient.SqlClient)<{ + readonly status: string; + readonly disposition: string; + readonly dropped_events: number; + readonly terminal_notices: number; + }>` + SELECT + exchange.status, + json_extract(dropped.payload, '$.disposition') AS disposition, + ( + SELECT COUNT(*) + FROM j5_a2a_comm_event + WHERE kind = 'exchange.dropped' AND exchange_id = ${opened.exchangeId} + ) AS dropped_events, + ( + SELECT COUNT(*) + FROM j5_a2a_delivery + WHERE exchange_role = 'terminal_notice' AND exchange_id = ${opened.exchangeId} + ) AS terminal_notices + FROM j5_a2a_exchange AS exchange + JOIN j5_a2a_comm_event AS dropped + ON dropped.exchange_id = exchange.exchange_id + AND dropped.kind = 'exchange.dropped' + WHERE exchange.exchange_id = ${opened.exchangeId} + `; + assert.deepStrictEqual(state, [ + { + status: "dropped", + disposition: "receiver-retired", + dropped_events: 1, + terminal_notices: 1, + }, + ]); + }).pipe(Effect.provide(makeTestLayer(notices, storedEvents))); + }), +); diff --git a/apps/server/src/j5/a2a/LifecycleService.ts b/apps/server/src/j5/a2a/LifecycleService.ts new file mode 100644 index 000000000000..dc7ce1380d31 --- /dev/null +++ b/apps/server/src/j5/a2a/LifecycleService.ts @@ -0,0 +1,466 @@ +import type { OrchestrationV2StoredEvent } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import type { SqlError } from "effect/unstable/sql/SqlError"; + +import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; +import { A2ADeliveryWorker } from "./DeliveryWorker.ts"; +import { + type ArchiveParticipantInput, + CommCommandId, + CorrelationId, + SquadronId, + ExchangeId, + type ExchangeDropDisposition, + isHumanParticipantId, + LIFECYCLE_PARTICIPANT_ID, + LedgerMessageId, + type LifecycleArchiveResult, + Participant, + ParticipantId, +} from "./contracts.ts"; +import { resolveThreadHome } from "./HomeRegistrar.ts"; +import { A2ALedger, type A2ALedgerError } from "./LedgerService.ts"; + +export class A2ALifecycleParticipantNotFoundError extends Schema.TaggedErrorClass()( + "A2ALifecycleParticipantNotFoundError", + { participantId: Schema.String }, +) { + override get message(): string { + return `Participant ${this.participantId} does not exist and cannot be archived.`; + } +} + +export class A2ALifecycleHumanArchiveNotAllowedError extends Schema.TaggedErrorClass()( + "A2ALifecycleHumanArchiveNotAllowedError", + { participantId: Schema.String }, +) { + override get message(): string { + return `Participant ${this.participantId} is a person. This operation archives agents only.`; + } +} + +export class A2ALifecycleCounterpartyStateError extends Schema.TaggedErrorClass()( + "A2ALifecycleCounterpartyStateError", + { + participantId: Schema.String, + exchangeId: Schema.String, + }, +) { + override get message(): string { + return `Cannot close exchange ${this.exchangeId}: affected participant ${this.participantId} has no readable Squadron projection.`; + } +} + +export class A2ALifecycleParticipantHomeStateError extends Schema.TaggedErrorClass()( + "A2ALifecycleParticipantHomeStateError", + { + participantId: Schema.String, + squadronIds: Schema.Array(Schema.String), + }, +) { + override get message(): string { + return `Participant ${this.participantId} has ambiguous immutable home history (${this.squadronIds.join(", ")}). Repair history before lifecycle retirement resumes.`; + } +} + +export class A2ALifecycleBridgeError extends Schema.TaggedErrorClass()( + "A2ALifecycleBridgeError", + { operation: Schema.String, cause: Schema.Defect() }, +) {} + +export type A2ALifecycleError = + | A2ALedgerError + | Schema.SchemaError + | SqlError + | A2ALifecycleParticipantNotFoundError + | A2ALifecycleHumanArchiveNotAllowedError + | A2ALifecycleCounterpartyStateError + | A2ALifecycleParticipantHomeStateError; + +export interface A2ALifecycleServiceShape { + /** Platform lifecycle authority only. This service is deliberately absent from the agent MCP toolkit. */ + readonly archiveParticipant: ( + input: ArchiveParticipantInput, + ) => Effect.Effect; + readonly handleStoredEvent: ( + event: OrchestrationV2StoredEvent, + ) => Effect.Effect; + readonly replayCommittedEvents: Effect.Effect; +} + +export class A2ALifecycleService extends Context.Service< + A2ALifecycleService, + A2ALifecycleServiceShape +>()("t3/j5/a2a/LifecycleService/A2ALifecycleService") {} + +interface MembershipRow { + readonly squadron_id: string; + readonly participant_id: string; + readonly participant_kind: "agent" | "human"; + readonly payload: string; +} + +interface HistoricalParticipantRow { + readonly squadron_id: string; + readonly payload: string; + readonly retired: number; +} + +interface ExchangeRow { + readonly squadron_id: string; + readonly exchange_id: string; + readonly sender_id: string; + readonly receiver_id: string; +} + +interface CursorRow { + readonly after_sequence: number; +} + +const stablePart = (value: string) => encodeURIComponent(value); + +const lifecycleKey = (exchange: ExchangeRow, disposition: ExchangeDropDisposition) => + `${stablePart(exchange.squadron_id)}:${stablePart(exchange.exchange_id)}:${disposition}`; + +const dropCommandId = (exchange: ExchangeRow, disposition: ExchangeDropDisposition) => + CommCommandId.make(`command:j5:a2a:lifecycle:drop:${lifecycleKey(exchange, disposition)}`); + +const noticeMessageId = (exchange: ExchangeRow, disposition: ExchangeDropDisposition) => + LedgerMessageId.make(`message:j5:a2a:lifecycle:drop:${lifecycleKey(exchange, disposition)}`); + +const noticeCorrelationId = (exchange: ExchangeRow, disposition: ExchangeDropDisposition) => + CorrelationId.make(`correlation:j5:a2a:lifecycle:drop:${lifecycleKey(exchange, disposition)}`); + +const participantArchiveCommandId = (squadronId: SquadronId, participantId: ParticipantId) => + CommCommandId.make( + `command:j5:a2a:lifecycle:participant:${stablePart(squadronId)}:${stablePart(participantId)}`, + ); + +export const formatLifecycleNotice = (input: { + readonly exchangeId: ExchangeId; + readonly retiredParticipantId: ParticipantId; + readonly disposition: ExchangeDropDisposition; +}): string => { + const consequence = + input.disposition === "receiver-retired" + ? "The receiver can no longer answer. Do not retry this exchange and do not replace the retired participant." + : "The asker is gone. Your reply obligation has ended; do not send a replacement reply."; + return [ + `[Cross-agent messaging system notice: exchange dropped]`, + `Exchange ${input.exchangeId} ended because ${input.retiredParticipantId} was retired from A2A (${input.disposition}).`, + consequence, + "Facts: replyRequired=false; retryAllowed=false; replacementRequired=false.", + "This is a platform-authored terminal notice, not a peer reply.", + ].join("\n\n"); +}; + +const decodeParticipant = Schema.decodeUnknownEffect(Schema.fromJsonString(Participant)); + +const bridgeError = (operation: string) => (cause: unknown) => + new A2ALifecycleBridgeError({ operation, cause }); + +const makeLayer = (daemon: boolean) => + Layer.effect( + A2ALifecycleService, + Effect.gen(function* () { + const ledger = yield* A2ALedger; + const worker = yield* A2ADeliveryWorker; + const threads = yield* ThreadManagement.ThreadManagementService; + const sql = yield* SqlClient.SqlClient; + const lifecyclePermit = yield* Semaphore.make(1); + + const membershipRows = Effect.fn("j5.a2a.lifecycle.membershipRows")(function* ( + participantId: ParticipantId, + ) { + return yield* sql` + SELECT squadron_id, participant_id, participant_kind, payload + FROM j5_a2a_squadron_membership + WHERE participant_id = ${participantId} + ORDER BY squadron_id + `; + }); + + const historicalParticipantRows = Effect.fn("j5.a2a.lifecycle.historicalParticipantRows")( + function* (participantId: ParticipantId) { + return yield* sql` + SELECT + joined.squadron_id, + json_extract(joined.payload, '$.participant') AS payload, + EXISTS ( + SELECT 1 + FROM j5_a2a_comm_event AS retirement + WHERE retirement.squadron_id = joined.squadron_id + AND retirement.seq > joined.seq + AND retirement.kind = 'participant.left' + AND json_extract(retirement.payload, '$.participant.kind') = 'agent' + AND json_extract(retirement.payload, '$.participant.id') = ${participantId} + AND json_extract(retirement.payload, '$.participant.threadId') = + json_extract(joined.payload, '$.participant.threadId') + ) AS retired + FROM j5_a2a_comm_event AS joined + WHERE joined.kind = 'participant.joined' + AND json_extract(joined.payload, '$.participant.id') = ${participantId} + ORDER BY joined.seq + LIMIT 2 + `; + }, + ); + + const counterpartySquadron = Effect.fn("j5.a2a.lifecycle.counterpartySquadron")(function* ( + participantId: ParticipantId, + exchange: ExchangeRow, + ) { + if (isHumanParticipantId(participantId)) { + return SquadronId.make(exchange.squadron_id); + } + const rows = yield* membershipRows(participantId); + const row = + rows.find((candidate) => candidate.squadron_id === exchange.squadron_id) ?? rows[0]; + if (row !== undefined) return SquadronId.make(row.squadron_id); + const historical = yield* historicalParticipantRows(participantId); + const historicalRow = + historical.find((candidate) => candidate.squadron_id === exchange.squadron_id) ?? + historical[0]; + if (historicalRow === undefined) { + return yield* new A2ALifecycleCounterpartyStateError({ + participantId, + exchangeId: exchange.exchange_id, + }); + } + return SquadronId.make(historicalRow.squadron_id); + }); + + const dropParticipantExchanges = Effect.fn("j5.a2a.lifecycle.dropParticipantExchanges")( + function* (input: { + readonly participantId: ParticipantId; + readonly squadronId: SquadronId; + readonly archivedAt: string; + }) { + const exchanges = yield* sql` + SELECT squadron_id, exchange_id, sender_id, receiver_id + FROM j5_a2a_exchange + WHERE status = 'open' + AND (sender_id = ${input.participantId} OR receiver_id = ${input.participantId}) + ORDER BY squadron_id, opened_seq, exchange_id + `; + const dropped: Array = []; + for (const exchange of exchanges) { + const disposition: ExchangeDropDisposition = + exchange.receiver_id === input.participantId ? "receiver-retired" : "sender-retired"; + const affectedParticipantId = ParticipantId.make( + disposition === "receiver-retired" ? exchange.sender_id : exchange.receiver_id, + ); + const exchangeId = ExchangeId.make(exchange.exchange_id); + const messageId = noticeMessageId(exchange, disposition); + const correlationId = noticeCorrelationId(exchange, disposition); + const receiverSquadronId = yield* counterpartySquadron(affectedParticipantId, exchange); + yield* ledger.appendEvents({ + commandId: dropCommandId(exchange, disposition), + squadronId: SquadronId.make(exchange.squadron_id), + acceptedAt: input.archivedAt, + events: [ + { + kind: "exchange.dropped", + sender: ParticipantId.make(exchange.sender_id), + receiver: ParticipantId.make(exchange.receiver_id), + exchangeId, + correlationId, + payload: { + disposition, + cause: { + kind: "participant-archived", + participantId: input.participantId, + squadronId: input.squadronId, + }, + facts: { + replyRequired: false, + retryAllowed: false, + replacementRequired: false, + }, + noticeMessageId: messageId, + }, + createdAt: input.archivedAt, + }, + { + kind: "message.sent", + sender: LIFECYCLE_PARTICIPANT_ID, + receiver: affectedParticipantId, + exchangeId, + correlationId, + payload: { + messageId, + text: formatLifecycleNotice({ + exchangeId, + retiredParticipantId: input.participantId, + disposition, + }), + originSquadronId: SquadronId.make(exchange.squadron_id), + receiverSquadronId, + exchangeRole: "terminal_notice", + envelopeChannel: "lifecycle_notice", + }, + createdAt: input.archivedAt, + }, + ], + }); + dropped.push(exchangeId); + } + return dropped; + }, + ); + + const archiveParticipantInternal = Effect.fn("j5.a2a.lifecycle.archiveParticipantInternal")( + function* (input: ArchiveParticipantInput) { + const rows = yield* historicalParticipantRows(input.participantId); + if (rows.length === 0) { + return yield* new A2ALifecycleParticipantNotFoundError({ + participantId: input.participantId, + }); + } + if (rows.length !== 1) { + return yield* new A2ALifecycleParticipantHomeStateError({ + participantId: input.participantId, + squadronIds: rows.map((row) => row.squadron_id), + }); + } + const row = rows[0]!; + const squadronId = SquadronId.make(row.squadron_id); + const participant = yield* decodeParticipant(row.payload); + if (participant.kind !== "agent") { + return yield* new A2ALifecycleHumanArchiveNotAllowedError({ + participantId: input.participantId, + }); + } + const archived = row.retired === 0; + yield* ledger.append({ + commandId: participantArchiveCommandId(squadronId, input.participantId), + squadronId, + acceptedAt: input.archivedAt, + event: { + kind: "participant.left", + sender: null, + receiver: input.participantId, + exchangeId: null, + correlationId: null, + payload: { participant }, + createdAt: input.archivedAt, + }, + }); + const droppedExchangeIds = yield* dropParticipantExchanges({ + participantId: input.participantId, + squadronId, + archivedAt: input.archivedAt, + }); + return { archived, droppedExchangeIds } satisfies LifecycleArchiveResult; + }, + ); + + const archiveParticipantRaw = Effect.fn("j5.a2a.lifecycle.archiveParticipant")(function* ( + input: ArchiveParticipantInput, + ) { + const result = yield* archiveParticipantInternal(input); + yield* worker.notify; + return result; + }); + const archiveParticipant: A2ALifecycleServiceShape["archiveParticipant"] = (input) => + lifecyclePermit.withPermit(archiveParticipantRaw(input)); + + const handleStoredEventRaw = Effect.fn("j5.a2a.lifecycle.handleStoredEvent")(function* ( + stored: OrchestrationV2StoredEvent, + ) { + if (stored.event.type !== "thread.archived" && stored.event.type !== "thread.deleted") { + return false; + } + const resolution = yield* resolveThreadHome(sql, stored.event.threadId).pipe( + Effect.catchTag("A2AHomeNotFoundError", () => Effect.succeed(null)), + ); + if (resolution === null) return false; + yield* archiveParticipant({ + participantId: resolution.home.participantId, + archivedAt: DateTime.formatIso(stored.event.occurredAt), + }); + return true; + }); + + const readCursor = Effect.fn("j5.a2a.lifecycle.readCursor")(function* () { + const rows = yield* sql` + SELECT after_sequence + FROM j5_a2a_lifecycle_cursor + WHERE singleton = 1 + `; + return rows[0]?.after_sequence ?? 0; + }); + + const writeCursor = Effect.fn("j5.a2a.lifecycle.writeCursor")(function* (sequence: number) { + const updatedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + yield* sql` + UPDATE j5_a2a_lifecycle_cursor + SET after_sequence = ${sequence}, updated_at = ${updatedAt} + WHERE singleton = 1 AND after_sequence < ${sequence} + `; + }); + + const replayCommittedEventsRaw = Effect.fn("j5.a2a.lifecycle.replayCommittedEvents")( + function* () { + let checkpoint = yield* readCursor(); + let lastSeen = checkpoint; + yield* threads.streamStoredEventsFrom({ afterSequence: checkpoint }).pipe( + Stream.runForEach((event) => + handleStoredEventRaw(event).pipe( + Effect.andThen((archived) => + Effect.gen(function* () { + lastSeen = event.sequence; + if (archived || event.sequence - checkpoint >= 128) { + yield* writeCursor(event.sequence); + checkpoint = event.sequence; + } + }), + ), + ), + ), + ); + if (lastSeen > checkpoint) yield* writeCursor(lastSeen); + }, + ); + + const handleStoredEvent: A2ALifecycleServiceShape["handleStoredEvent"] = (event) => + handleStoredEventRaw(event).pipe(Effect.mapError(bridgeError("handle thread retirement"))); + const replayCommittedEvents: A2ALifecycleServiceShape["replayCommittedEvents"] = + replayCommittedEventsRaw().pipe(Effect.mapError(bridgeError("replay thread retirements"))); + + if (daemon) { + let retryDelayMs = 250; + const run = Effect.forever( + replayCommittedEventsRaw().pipe( + Effect.andThen(Effect.die("J5 A2A lifecycle retirement stream ended")), + Effect.catchCause((cause) => { + const delayMs = retryDelayMs; + retryDelayMs = Math.min(delayMs * 2, 30_000); + return Effect.logWarning( + "J5 A2A lifecycle retirement stream failed; resuming from cursor", + { cause, retryDelayMs: delayMs }, + ).pipe(Effect.andThen(Effect.sleep(Duration.millis(delayMs)))); + }), + ), + ); + yield* Effect.forkScoped(run); + } + + return A2ALifecycleService.of({ + archiveParticipant, + handleStoredEvent, + replayCommittedEvents, + }); + }), + ); + +export const manualLayer = makeLayer(false); +export const layer = makeLayer(true); diff --git a/apps/server/src/j5/a2a/Migrations.test.ts b/apps/server/src/j5/a2a/Migrations.test.ts index a7d4415a37d0..eb88fd60e5a3 100644 --- a/apps/server/src/j5/a2a/Migrations.test.ts +++ b/apps/server/src/j5/a2a/Migrations.test.ts @@ -10,6 +10,7 @@ import { } from "../../persistence/Migrations.ts"; import { J5_A2A_MIGRATIONS_TABLE, migrationEntries, runJ5A2AMigrations } from "./Migrations.ts"; import Migration0005 from "./migrations/005_ImmutableThreadHome.ts"; +import Migration0008 from "./migrations/008_LifecycleClosure.ts"; const enableAndAssertForeignKeys = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -46,6 +47,7 @@ it.effect("tracks J5 A2A migrations independently from upstream migrations", () { migration_id: 5, name: "ImmutableThreadHome" }, { migration_id: 6, name: "HumanNode" }, { migration_id: 7, name: "ParticipantPlacement" }, + { migration_id: 8, name: "LifecycleClosure" }, ]); assert.deepStrictEqual( migrationEntries.map(([id, name]) => [id, name]), @@ -57,6 +59,7 @@ it.effect("tracks J5 A2A migrations independently from upstream migrations", () [5, "ImmutableThreadHome"], [6, "HumanNode"], [7, "ParticipantPlacement"], + [8, "LifecycleClosure"], ], ); }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), @@ -88,7 +91,8 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c 'j5_a2a_human_inbox_data', 'j5_a2a_silence_detector_cursor', 'j5_a2a_placement_event', - 'j5_a2a_participant_placement' + 'j5_a2a_participant_placement', + 'j5_a2a_lifecycle_cursor' ) ORDER BY name `; @@ -146,6 +150,7 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c { name: "j5_a2a_human_inbox" }, { name: "j5_a2a_human_inbox_data" }, { name: "j5_a2a_human_person" }, + { name: "j5_a2a_lifecycle_cursor" }, { name: "j5_a2a_participant_placement" }, { name: "j5_a2a_placement_event" }, { name: "j5_a2a_silence_detector_cursor" }, @@ -222,6 +227,13 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c SELECT after_sequence FROM j5_a2a_silence_detector_cursor WHERE singleton = 1 `; assert.deepStrictEqual(cursor, [{ after_sequence: null }]); + const lifecycleCursor = yield* sql<{ + readonly after_sequence: number; + readonly updated_at: string | null; + }>` + SELECT after_sequence, updated_at FROM j5_a2a_lifecycle_cursor WHERE singleton = 1 + `; + assert.deepStrictEqual(lifecycleCursor, [{ after_sequence: 0, updated_at: null }]); assert.deepStrictEqual(unprefixed, []); yield* sql` INSERT INTO j5_a2a_squadron (id, name, created_at) @@ -322,6 +334,259 @@ it.effect("requires a non-null, non-blank reparent actor subject", () => }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), ); +it.effect("adds lifecycle terminal state without mutating the A4 human inbox projection", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runJ5A2AMigrations({ toMigrationInclusive: 7 }); + yield* sql` + INSERT INTO j5_a2a_squadron (id, name, created_at) + VALUES ('squadron:lifecycle-migration', 'Lifecycle migration', '2026-08-23T00:00:00.000Z') + `; + yield* sql` + INSERT INTO j5_a2a_human_person (person_id, is_local_operator, created_at) + VALUES ('human:person-lifecycle-migration', 1, '2026-08-23T00:00:00.000Z') + `; + yield* sql` + INSERT INTO j5_a2a_exchange ( + squadron_id, exchange_id, sender_id, receiver_id, status, intent, urgency, + opened_seq, closed_seq, created_at, updated_at + ) VALUES + ( + 'squadron:lifecycle-migration', 'exchange:lifecycle:open', + 'agent:lifecycle-migration', 'human:person-lifecycle-migration', 'open', + 'Preserve open inbox state', 'soon', 1, NULL, + '2026-08-23T00:00:00.000Z', '2026-08-23T00:00:00.000Z' + ), + ( + 'squadron:lifecycle-migration', 'exchange:lifecycle:answered', + 'agent:lifecycle-migration', 'human:person-lifecycle-migration', 'closed', + 'Preserve answered inbox state', 'blocking', 2, 5, + '2026-08-23T00:00:01.000Z', '2026-08-23T00:00:05.000Z' + ), + ( + 'squadron:lifecycle-migration', 'exchange:lifecycle:dropped', + 'agent:lifecycle-migration', 'human:person-lifecycle-migration', 'closed', + 'Preserve dropped inbox state', 'fyi', 3, 6, + '2026-08-23T00:00:02.000Z', '2026-08-23T00:00:06.000Z' + ) + `; + yield* sql` + INSERT INTO j5_a2a_human_inbox_data ( + origin_squadron_id, message_id, exchange_id, sender_id, receiver_id, payload, created_at + ) VALUES ( + 'squadron:lifecycle-migration', 'message:raw-inbox', 'exchange:lifecycle:open', + 'agent:lifecycle-migration', 'human:person-lifecycle-migration', + '{"opaque":"raw-inbox"}', '2026-08-23T00:00:00.000Z' + ) + `; + yield* sql` + INSERT INTO j5_a2a_delivery ( + squadron_id, message_id, command_id, sent_seq, sender_id, receiver_id, + receiver_squadron_id, exchange_id, exchange_role, correlation_id, + message_text, status, attempts, last_error, next_attempt_at, delivered_seq, + created_at, updated_at, envelope_channel + ) VALUES ( + 'squadron:lifecycle-migration', 'message:lifecycle-migration', + 'command:lifecycle-migration', 4, 'agent:lifecycle-migration', + 'human:person-lifecycle-migration', 'squadron:lifecycle-migration', + 'exchange:lifecycle:open', 'ask', 'correlation:lifecycle-migration', + 'Preserve this delivery', 'pending', 0, NULL, NULL, NULL, + '2026-08-23T00:00:00.000Z', '2026-08-23T00:00:00.000Z', 'peer' + ) + `; + yield* sql` + INSERT INTO j5_a2a_human_inbox ( + person_id, squadron_id, exchange_id, sender_id, intent, urgency, + latest_message_id, latest_message, opened_seq, opened_at, status, + terminal_seq, terminal_at, terminal_disposition, terminal_cause, + terminal_facts, terminal_notice_message_id + ) VALUES + ( + 'human:person-lifecycle-migration', 'squadron:lifecycle-migration', + 'exchange:lifecycle:open', 'agent:lifecycle-migration', 'Open intent', 'soon', + 'message:open', 'Open message', 1, '2026-08-23T00:00:00.000Z', 'open', + NULL, NULL, NULL, NULL, NULL, NULL + ), + ( + 'human:person-lifecycle-migration', 'squadron:lifecycle-migration', + 'exchange:lifecycle:answered', 'agent:lifecycle-migration', 'Answered intent', + 'blocking', 'message:answered', 'Answered message', 2, + '2026-08-23T00:00:01.000Z', 'answered', 5, '2026-08-23T00:00:05.000Z', + 'reply-received', '{"kind":"reply"}', '{"replyRequired":false}', NULL + ), + ( + 'human:person-lifecycle-migration', 'squadron:lifecycle-migration', + 'exchange:lifecycle:dropped', 'agent:lifecycle-migration', 'Dropped intent', 'fyi', + 'message:dropped', 'Dropped message', 3, '2026-08-23T00:00:02.000Z', + 'dropped', 6, '2026-08-23T00:00:06.000Z', 'sender-retired', + '{"kind":"participant-archived"}', + '{"replyRequired":false,"retryAllowed":false,"replacementRequired":false}', + 'message:terminal-notice' + ) + `; + const inboxSqlBefore = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'j5_a2a_human_inbox' + `; + const inboxRowsBefore = yield* sql>>` + SELECT * FROM j5_a2a_human_inbox ORDER BY exchange_id + `; + const rawInboxRowsBefore = yield* sql>>` + SELECT * FROM j5_a2a_human_inbox_data ORDER BY message_id + `; + + yield* sql.withTransaction(Migration0008); + + const inboxSqlAfter = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'j5_a2a_human_inbox' + `; + const inboxRowsAfter = yield* sql>>` + SELECT * FROM j5_a2a_human_inbox ORDER BY exchange_id + `; + const rawInboxRowsAfter = yield* sql>>` + SELECT * FROM j5_a2a_human_inbox_data ORDER BY message_id + `; + const foreignKeyViolations = yield* sql>>` + PRAGMA foreign_key_check + `; + const exchangeColumns = yield* sql<{ readonly name: string; readonly pk: number }>` + PRAGMA table_info(j5_a2a_exchange) + `; + const exchangeForeignKeys = yield* sql<{ + readonly table: string; + readonly from: string; + readonly to: string; + readonly on_delete: string; + }>`PRAGMA foreign_key_list(j5_a2a_exchange)`; + const inboxForeignKeys = yield* sql<{ + readonly table: string; + readonly from: string; + readonly to: string; + readonly on_delete: string; + }>`PRAGMA foreign_key_list(j5_a2a_human_inbox)`; + const squadronColumns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(j5_a2a_squadron) + `; + const exchangeSchema = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'j5_a2a_exchange' + `; + const openPairIndex = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master + WHERE type = 'index' AND name = 'j5_a2a_exchange_open_pair_idx' + `; + const deliverySchema = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'j5_a2a_delivery' + `; + const eventSchema = yield* sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'j5_a2a_comm_event' + `; + const preservedDelivery = yield* sql<{ + readonly message_id: string; + readonly exchange_role: string; + readonly envelope_channel: string; + }>` + SELECT message_id, exchange_role, envelope_channel + FROM j5_a2a_delivery + WHERE message_id = 'message:lifecycle-migration' + `; + + assert.deepStrictEqual(inboxSqlAfter, inboxSqlBefore); + assert.include( + inboxSqlAfter[0]?.sql ?? "", + "REFERENCES j5_a2a_exchange(squadron_id, exchange_id) ON DELETE CASCADE", + ); + assert.deepStrictEqual(inboxRowsAfter, inboxRowsBefore); + assert.deepStrictEqual(rawInboxRowsAfter, rawInboxRowsBefore); + assert.deepStrictEqual(foreignKeyViolations, []); + assert.deepStrictEqual( + exchangeColumns.filter((column) => column.pk > 0).map(({ name, pk }) => ({ name, pk })), + [ + { name: "squadron_id", pk: 1 }, + { name: "exchange_id", pk: 2 }, + ], + ); + assert.isTrue( + exchangeForeignKeys.some( + (foreignKey) => + foreignKey.table === "j5_a2a_squadron" && + foreignKey.from === "squadron_id" && + foreignKey.to === "id" && + foreignKey.on_delete === "CASCADE", + ), + ); + assert.isTrue( + inboxForeignKeys.some( + (foreignKey) => + foreignKey.table === "j5_a2a_exchange" && + foreignKey.from === "squadron_id" && + foreignKey.to === "squadron_id" && + foreignKey.on_delete === "CASCADE", + ), + ); + assert.isTrue( + inboxForeignKeys.some( + (foreignKey) => + foreignKey.table === "j5_a2a_exchange" && + foreignKey.from === "exchange_id" && + foreignKey.to === "exchange_id" && + foreignKey.on_delete === "CASCADE", + ), + ); + assert.notInclude( + squadronColumns.map((column) => column.name), + "archived_at", + ); + assert.match( + exchangeSchema[0]?.sql ?? "", + /status TEXT NOT NULL CHECK \(status IN \('open', 'closed', 'dropped'\)\)/, + ); + assert.match( + openPairIndex[0]?.sql ?? "", + /CREATE UNIQUE INDEX j5_a2a_exchange_open_pair_idx[\s\S]*WHERE status = 'open'/, + ); + assert.include(deliverySchema[0]?.sql ?? "", "'terminal_notice'"); + assert.include(deliverySchema[0]?.sql ?? "", "'lifecycle_notice'"); + assert.include(eventSchema[0]?.sql ?? "", "'exchange.dropped'"); + assert.notInclude(eventSchema[0]?.sql ?? "", "'squadron.archived'"); + assert.deepStrictEqual(preservedDelivery, [ + { + message_id: "message:lifecycle-migration", + exchange_role: "ask", + envelope_channel: "peer", + }, + ]); + + yield* sql` + UPDATE j5_a2a_exchange + SET status = 'dropped', closed_seq = 7, updated_at = '2026-08-23T00:01:00.000Z' + WHERE squadron_id = 'squadron:lifecycle-migration' + AND exchange_id = 'exchange:lifecycle:dropped' + `; + const invalidStatusError = yield* Effect.flip(sql` + UPDATE j5_a2a_exchange + SET status = 'invalid-terminal-state' + WHERE squadron_id = 'squadron:lifecycle-migration' + AND exchange_id = 'exchange:lifecycle:open' + `); + assert.equal(invalidStatusError._tag, "SqlError"); + yield* sql` + INSERT INTO j5_a2a_delivery ( + squadron_id, message_id, command_id, sent_seq, sender_id, receiver_id, + receiver_squadron_id, exchange_id, exchange_role, correlation_id, + message_text, status, attempts, last_error, next_attempt_at, delivered_seq, + created_at, updated_at, envelope_channel + ) VALUES ( + 'squadron:lifecycle-migration', 'message:lifecycle-notice-migration', + 'command:lifecycle-notice-migration', 8, 'platform:lifecycle', + 'human:person-lifecycle-migration', 'squadron:lifecycle-migration', + 'exchange:lifecycle:dropped', 'terminal_notice', + 'correlation:lifecycle-notice-migration', 'Exchange dropped', 'pending', 0, + NULL, NULL, NULL, '2026-08-23T00:01:00.000Z', + '2026-08-23T00:01:00.000Z', 'lifecycle_notice' + ) + `; + }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), +); + it.effect("reports conflicting thread ids before creating the immutable-home index", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/src/j5/a2a/Migrations.ts b/apps/server/src/j5/a2a/Migrations.ts index b54d0b7b703f..0b8f62493464 100644 --- a/apps/server/src/j5/a2a/Migrations.ts +++ b/apps/server/src/j5/a2a/Migrations.ts @@ -8,10 +8,11 @@ import Migration0004 from "./migrations/004_SilenceNoticeChannel.ts"; import Migration0005 from "./migrations/005_ImmutableThreadHome.ts"; import Migration0006 from "./migrations/006_HumanNode.ts"; import Migration0007 from "./migrations/007_ParticipantPlacement.ts"; +import Migration0008 from "./migrations/008_LifecycleClosure.ts"; export const J5_A2A_MIGRATIONS_TABLE = "j5_a2a_migrations"; -// Entries 1 through 6 are persisted migration history. Keep their file and manifest names unchanged; +// Entries 1 through 8 are persisted migration history. Keep their file and manifest names unchanged; // applied migrations are skipped by id and never rerun. export const migrationEntries = [ [1, "EpicCommunicationLedger", Migration0001], @@ -21,6 +22,7 @@ export const migrationEntries = [ [5, "ImmutableThreadHome", Migration0005], [6, "HumanNode", Migration0006], [7, "ParticipantPlacement", Migration0007], + [8, "LifecycleClosure", Migration0008], ] as const; const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/j5/a2a/README.md b/apps/server/src/j5/a2a/README.md index 3ec29f5ba54c..1ac154e2de3c 100644 --- a/apps/server/src/j5/a2a/README.md +++ b/apps/server/src/j5/a2a/README.md @@ -21,3 +21,13 @@ Byte-equivalent rebuilds for the A2 exchange and delivery projections are deferr A3 requires every sent-message event and delivery row to declare an explicit envelope channel. Peer sends write `peer`; silence notices write `silence_notice`, reuse A2's versioned silence-notice formatter, and bypass peer/human wrapping in the shared transport. The silence detector checkpoints its global v2 event sequence in the J5 migration lane. A new installation first reconciles only open delivered exchanges against current thread projections, records the existing v2 high-water mark, and then tails from that durable cursor; retries replay at most the uncheckpointed tail instead of scanning the event store from genesis on every boot. A broken v2 subscription uses capped exponential backoff without re-running the reconciliation scan on every attempt. + +Archive closure is a separate platform lifecycle write authority in `LifecycleService.ts`; it is never an A3 silence action and is not exposed through the agent MCP toolkit. Participant archive appends `participant.left` with the complete original agent identity, then terminates every affected open exchange with `exchange.dropped` plus one stable platform-authored `lifecycle_notice` delivery. The existing membership projector removes retired participants from active reads while the append-only ledger remains readable indefinitely. + +Squadron archive behavior and its user surface are deferred to the SC4 polished-later archive flow. This layer deliberately has no Squadron archive command, event, projection column, HTTP route, MCP tool, or warning/confirmation path; that future feature can compose the participant termination machinery once its product contract ships. + +The bounded lifecycle reactor consumes committed `thread.archived` and `thread.deleted` events from a durable cursor and applies that same idempotent termination authority. Native threads without an A2A home are explicit no-ops. A later upstream `thread.unarchived` event does not revive the retired A2A participant: dropped exchanges and notices remain historical facts, and participation cannot resume until a separately ratified re-entry lifecycle exists. + +`ArchiveFactsService.ts` is the read-only AR2 boundary for a future human pre-archive dialog. It reports open exchanges in both directions without deciding archive or settle policy. Placement is supplied through a provider whose current production result is explicitly `unknown`; a checked empty subtree is a different `none` state. HTTP registration on the sanctioned A4 routes layer remains deferred, and the measured MCP option is rejected because an agent tool is the wrong consumer for a human archive dialog. + +The future A2S `archive_agent` operation is expected to compose these public service boundaries directly: `ArchiveFactsService` supplies the AR2 preflight read and `LifecycleService.archiveParticipant` supplies the same idempotent R1 termination engine used by the lifecycle reactor. A9 does not ship an `archive_agent` handler or tool stub. diff --git a/apps/server/src/j5/a2a/SendService.ts b/apps/server/src/j5/a2a/SendService.ts index adb6272bfa9c..753dac14642d 100644 --- a/apps/server/src/j5/a2a/SendService.ts +++ b/apps/server/src/j5/a2a/SendService.ts @@ -80,6 +80,18 @@ export class A2AAmbiguousParticipantError extends Schema.TaggedErrorClass()( + "A2AParticipantArchivedError", + { + participantId: Schema.String, + squadronId: Schema.String, + }, +) { + override get message(): string { + return `Participant ${this.participantId} is A2A-retired from immutable home ${this.squadronId} by participant.left and cannot receive new messages. Upstream thread unarchive does not revive A2A participation; choose an active participant or wait for a ratified re-entry policy.`; + } +} + export class A2AIntentRequiredError extends Schema.TaggedErrorClass()( "A2AIntentRequiredError", {}, @@ -156,6 +168,7 @@ export type A2ASendError = | A2AHomeMembershipStateError | A2AParticipantNotFoundError | A2AAmbiguousParticipantError + | A2AParticipantArchivedError | A2AIntentRequiredError | A2AUrgencyRequiredError | A2AUrgencyNotAcceptedError @@ -167,16 +180,19 @@ export type A2ASendError = interface MembershipRow { readonly squadron_id: string; readonly participant_id: string; - readonly thread_id: string | null; readonly payload: string; } +interface RetiredParticipantRow { + readonly squadron_id: string; +} + interface ExchangeRow { readonly squadron_id: string; readonly exchange_id: string; readonly sender_id: string; readonly receiver_id: string; - readonly status: "open" | "closed"; + readonly status: "open" | "closed" | "dropped"; } interface ExistingMessageRow { @@ -184,7 +200,7 @@ interface ExistingMessageRow { readonly sender_id: string; readonly receiver_id: string; readonly exchange_id: string | null; - readonly exchange_role: "none" | "ask" | "followup" | "reply"; + readonly exchange_role: "none" | "ask" | "followup" | "reply" | "terminal_notice"; readonly sent_seq: number; } @@ -219,12 +235,37 @@ export const layer: Layer.Layer` - SELECT squadron_id, participant_id, thread_id, payload + SELECT squadron_id, participant_id, payload FROM j5_a2a_squadron_membership ORDER BY squadron_id, participant_id `; }); + const retiredParticipantRows = Effect.fn("j5.a2a.send.retiredParticipantRows")(function* ( + id: ParticipantId, + ) { + return yield* sql` + SELECT joined.squadron_id + FROM j5_a2a_comm_event AS joined + WHERE joined.kind = 'participant.joined' + AND json_extract(joined.payload, '$.participant.kind') = 'agent' + AND json_extract(joined.payload, '$.participant.id') = ${id} + AND EXISTS ( + SELECT 1 + FROM j5_a2a_comm_event AS retirement + WHERE retirement.squadron_id = joined.squadron_id + AND retirement.seq > joined.seq + AND retirement.kind = 'participant.left' + AND json_extract(retirement.payload, '$.participant.kind') = 'agent' + AND json_extract(retirement.payload, '$.participant.id') = ${id} + AND json_extract(retirement.payload, '$.participant.threadId') = + json_extract(joined.payload, '$.participant.threadId') + ) + ORDER BY joined.seq + LIMIT 2 + `; + }); + const senderMembership = Effect.fn("j5.a2a.send.senderMembership")(function* ( threadId: ThreadId, ) { @@ -277,7 +318,17 @@ export const layer: Layer.Layer row.participant_id === id); if (matches.length === 0) { - return yield* new A2AParticipantNotFoundError({ participantId: id }); + const retired = yield* retiredParticipantRows(id); + if (retired[0] === undefined) { + return yield* new A2AParticipantNotFoundError({ participantId: id }); + } + if (retired.length > 1) { + return yield* new A2AAmbiguousParticipantError({ participantId: id }); + } + return yield* new A2AParticipantArchivedError({ + participantId: id, + squadronId: retired[0].squadron_id, + }); } if (matches.length > 1) { return yield* new A2AAmbiguousParticipantError({ participantId: id }); diff --git a/apps/server/src/j5/a2a/SilenceDetector.test.ts b/apps/server/src/j5/a2a/SilenceDetector.test.ts index 8fe7232633a5..b33d312d7440 100644 --- a/apps/server/src/j5/a2a/SilenceDetector.test.ts +++ b/apps/server/src/j5/a2a/SilenceDetector.test.ts @@ -42,13 +42,13 @@ import { import { CommCommandId, CorrelationId, + ExchangeId, + LedgerMessageId, SquadronId, ParticipantId, SILENCE_DETECTOR_PARTICIPANT_ID, type AgentParticipant, type HumanParticipant, - type ExchangeId, - type LedgerMessageId, } from "./contracts.ts"; const squadronId = SquadronId.make("squadron:silence-test"); @@ -301,6 +301,87 @@ const markAlarmed = Effect.fn("test.j5.a2a.silence.markAlarmed")(function* ( }); }); +const silenceAppendCommand = ( + exchangeId: ExchangeId, + deliveryMessageId: LedgerMessageId, + suffix: string, +) => { + const observedAt = iso(6); + const correlationId = CorrelationId.make(`correlation:silence:serialization:${suffix}`); + return { + commandId: CommCommandId.make(`command:silence:serialization:${suffix}`), + squadronId, + acceptedAt: observedAt, + events: [ + { + kind: "silence.notice" as const, + sender: null, + receiver: waiter.id, + exchangeId, + correlationId, + payload: { + subjectId: subject.id, + deliveryMessageId, + observedAt, + state: "turn-ended-no-reply" as const, + runId: null, + processing: "never-processed" as const, + }, + createdAt: observedAt, + }, + { + kind: "message.sent" as const, + sender: SILENCE_DETECTOR_PARTICIPANT_ID, + receiver: waiter.id, + exchangeId: null, + correlationId, + payload: { + messageId: LedgerMessageId.make(`message:silence:serialization:${suffix}`), + text: "Serialization seam silence notice", + originSquadronId: squadronId, + receiverSquadronId: squadronId, + exchangeRole: "none" as const, + envelopeChannel: "silence_notice" as const, + }, + createdAt: observedAt, + }, + ], + }; +}; + +const dropExchange = Effect.fn("test.j5.a2a.silence.dropExchange")(function* ( + exchangeId: ExchangeId, + suffix: string, +) { + yield* (yield* A2ALedger).append({ + commandId: CommCommandId.make(`command:silence:serialization:drop:${suffix}`), + squadronId, + acceptedAt: iso(5), + event: { + kind: "exchange.dropped", + sender: waiter.id, + receiver: subject.id, + exchangeId, + correlationId: CorrelationId.make(`correlation:silence:serialization:drop:${suffix}`), + payload: { + disposition: "receiver-retired", + cause: { + kind: "participant-archived", + participantId: subject.id, + squadronId, + }, + facts: { + replyRequired: false, + retryAllowed: false, + replacementRequired: false, + }, + noticeMessageId: LedgerMessageId.make(`message:silence:serialization:drop:${suffix}`), + }, + createdAt: iso(5), + }, + }); +}); + const terminalEvent = ( status: "completed" | "failed" | "interrupted" | "cancelled" | "rolled_back", sequence = 100, @@ -473,6 +554,68 @@ it.effect("emits processed mid-turn silence and dedupes a later lifecycle sequen }).pipe(Effect.provide(makeTestLayer())), ); +it.effect("no-ops when A9 drops after the A3 read but before its serialized append", () => + Effect.gen(function* () { + yield* seed(); + const inbound = yield* seedInbound(2); + assert.isNotNull(inbound.exchangeId); + const exchangeId = inbound.exchangeId!; + const sql = yield* SqlClient.SqlClient; + const observed = yield* sql<{ readonly status: string }>` + SELECT status FROM j5_a2a_exchange WHERE exchange_id = ${exchangeId} + `; + assert.deepStrictEqual(observed, [{ status: "open" }]); + + yield* dropExchange(exchangeId, "drop-before-append"); + const attempted = yield* (yield* A2ALedger).appendEventsIfExchangeOpen( + silenceAppendCommand(exchangeId, inbound.messageId, "drop-before-append"), + exchangeId, + ); + assert.isNull(attempted); + + const rows = yield* sql<{ + readonly dropped: number; + readonly silence: number; + readonly silence_deliveries: number; + }>` + SELECT + (SELECT COUNT(*) FROM j5_a2a_comm_event WHERE kind = 'exchange.dropped' AND exchange_id = ${exchangeId}) AS dropped, + (SELECT COUNT(*) FROM j5_a2a_comm_event WHERE kind = 'silence.notice' AND exchange_id = ${exchangeId}) AS silence, + (SELECT COUNT(*) FROM j5_a2a_delivery WHERE envelope_channel = 'silence_notice' AND command_id = 'command:silence:serialization:drop-before-append') AS silence_deliveries + `; + assert.deepStrictEqual(rows, [{ dropped: 1, silence: 0, silence_deliveries: 0 }]); + }).pipe(Effect.provide(makeTestLayer())), +); + +it.effect("keeps both historical facts when A3 appends before A9 drops", () => + Effect.gen(function* () { + yield* seed(); + const inbound = yield* seedInbound(2); + assert.isNotNull(inbound.exchangeId); + const exchangeId = inbound.exchangeId!; + const ledger = yield* A2ALedger; + const appended = yield* ledger.appendEventsIfExchangeOpen( + silenceAppendCommand(exchangeId, inbound.messageId, "append-before-drop"), + exchangeId, + ); + assert.isNotNull(appended); + assert.isTrue(appended!.committed); + + yield* dropExchange(exchangeId, "append-before-drop"); + const rows = yield* (yield* SqlClient.SqlClient)<{ + readonly dropped: number; + readonly silence: number; + readonly silence_deliveries: number; + }>` + SELECT + (SELECT COUNT(*) FROM j5_a2a_comm_event WHERE kind = 'exchange.dropped' AND exchange_id = ${exchangeId}) AS dropped, + (SELECT COUNT(*) FROM j5_a2a_comm_event WHERE kind = 'silence.notice' AND exchange_id = ${exchangeId}) AS silence, + (SELECT COUNT(*) FROM j5_a2a_delivery WHERE envelope_channel = 'silence_notice' AND command_id = 'command:silence:serialization:append-before-drop') AS silence_deliveries + `; + assert.deepStrictEqual(rows, [{ dropped: 1, silence: 1, silence_deliveries: 1 }]); + }).pipe(Effect.provide(makeTestLayer())), +); + it.effect("emits never-processed when no turn started after the latest delivery", () => Effect.gen(function* () { yield* seed(); diff --git a/apps/server/src/j5/a2a/SilenceDetector.ts b/apps/server/src/j5/a2a/SilenceDetector.ts index efd927dabbad..5d0437eaeb66 100644 --- a/apps/server/src/j5/a2a/SilenceDetector.ts +++ b/apps/server/src/j5/a2a/SilenceDetector.ts @@ -359,45 +359,49 @@ const makeLayer = (daemon: boolean) => exchange.exchange_id, payload.deliveryMessageId, ); - const result = yield* ledger.appendEvents({ - commandId: commandIdFor( - exchange.squadron_id, - exchange.exchange_id, - payload.deliveryMessageId, - ), - squadronId: SquadronId.make(exchange.squadron_id), - acceptedAt: payload.observedAt, - events: [ - { - kind: "silence.notice", - sender: null, - receiver: ParticipantId.make(exchange.sender_id), - exchangeId, - correlationId, - payload, - createdAt: payload.observedAt, - }, - { - kind: "message.sent", - sender: SILENCE_DETECTOR_PARTICIPANT_ID, - receiver: ParticipantId.make(exchange.sender_id), - exchangeId: null, - correlationId, - payload: { - messageId, - text: formatSilenceNoticeEnvelope({ - noticeType: payload.state, - message: noticeMessage(payload, exchangeId), - }), - originSquadronId: SquadronId.make(exchange.squadron_id), - receiverSquadronId: SquadronId.make(exchange.squadron_id), - exchangeRole: "none", - envelopeChannel: "silence_notice", + const result = yield* ledger.appendEventsIfExchangeOpen( + { + commandId: commandIdFor( + exchange.squadron_id, + exchange.exchange_id, + payload.deliveryMessageId, + ), + squadronId: SquadronId.make(exchange.squadron_id), + acceptedAt: payload.observedAt, + events: [ + { + kind: "silence.notice", + sender: null, + receiver: ParticipantId.make(exchange.sender_id), + exchangeId, + correlationId, + payload, + createdAt: payload.observedAt, }, - createdAt: payload.observedAt, - }, - ], - }); + { + kind: "message.sent", + sender: SILENCE_DETECTOR_PARTICIPANT_ID, + receiver: ParticipantId.make(exchange.sender_id), + exchangeId: null, + correlationId, + payload: { + messageId, + text: formatSilenceNoticeEnvelope({ + noticeType: payload.state, + message: noticeMessage(payload, exchangeId), + }), + originSquadronId: SquadronId.make(exchange.squadron_id), + receiverSquadronId: SquadronId.make(exchange.squadron_id), + exchangeRole: "none", + envelopeChannel: "silence_notice", + }, + createdAt: payload.observedAt, + }, + ], + }, + exchangeId, + ); + if (result === null) return []; return result.committed ? result.events.filter((event) => event.kind === "silence.notice") : []; diff --git a/apps/server/src/j5/a2a/contracts.ts b/apps/server/src/j5/a2a/contracts.ts index 01e9dc0ee967..831652fb5b65 100644 --- a/apps/server/src/j5/a2a/contracts.ts +++ b/apps/server/src/j5/a2a/contracts.ts @@ -29,10 +29,15 @@ export type LedgerMessageId = typeof LedgerMessageId.Type; export const Urgency = Schema.Literals(["blocking", "soon", "fyi"]); export type Urgency = typeof Urgency.Type; -export const DeliveryEnvelopeChannel = Schema.Literals(["peer", "silence_notice"]); +export const DeliveryEnvelopeChannel = Schema.Literals([ + "peer", + "silence_notice", + "lifecycle_notice", +]); export type DeliveryEnvelopeChannel = typeof DeliveryEnvelopeChannel.Type; export const SILENCE_DETECTOR_PARTICIPANT_ID = ParticipantId.make("platform:silence-detector"); +export const LIFECYCLE_PARTICIPANT_ID = ParticipantId.make("platform:lifecycle"); export const isHumanParticipantId = (id: ParticipantId): boolean => id.startsWith("human:") && id.length > "human:".length; @@ -78,6 +83,7 @@ export const CommEventKind = Schema.Literals([ "message.delivered", "message.delivery_failed", "exchange.closed", + "exchange.dropped", "silence.notice", "participant.joined", "participant.left", @@ -90,6 +96,7 @@ const NonMembershipEventKind = Schema.Literals([ "message.delivered", "message.delivery_failed", "exchange.closed", + "exchange.dropped", "silence.notice", ]); @@ -191,7 +198,7 @@ export const MessageSentPayload = Schema.Struct({ text: Schema.String.check(Schema.isNonEmpty()), originSquadronId: SquadronId, receiverSquadronId: SquadronId, - exchangeRole: Schema.Literals(["none", "ask", "followup", "reply"]), + exchangeRole: Schema.Literals(["none", "ask", "followup", "reply", "terminal_notice"]), envelopeChannel: DeliveryEnvelopeChannel, }); export type MessageSentPayload = typeof MessageSentPayload.Type; @@ -217,6 +224,25 @@ export const ExchangeClosedPayload = Schema.Struct({ }); export type ExchangeClosedPayload = typeof ExchangeClosedPayload.Type; +export const ExchangeDropDisposition = Schema.Literals(["receiver-retired", "sender-retired"]); +export type ExchangeDropDisposition = typeof ExchangeDropDisposition.Type; + +export const ExchangeDroppedPayload = Schema.Struct({ + disposition: ExchangeDropDisposition, + cause: Schema.Struct({ + kind: Schema.Literal("participant-archived"), + participantId: ParticipantId, + squadronId: SquadronId, + }), + facts: Schema.Struct({ + replyRequired: Schema.Literal(false), + retryAllowed: Schema.Literal(false), + replacementRequired: Schema.Literal(false), + }), + noticeMessageId: LedgerMessageId, +}); +export type ExchangeDroppedPayload = typeof ExchangeDroppedPayload.Type; + export const SendMessageInput = Schema.Struct({ commandId: CommCommandId, senderThreadId: ThreadId, @@ -314,3 +340,15 @@ export const AnswerHumanExchangeInput = Schema.Struct({ acceptedAt: Schema.String, }); export type AnswerHumanExchangeInput = typeof AnswerHumanExchangeInput.Type; + +export const ArchiveParticipantInput = Schema.Struct({ + participantId: ParticipantId, + archivedAt: Schema.String, +}); +export type ArchiveParticipantInput = typeof ArchiveParticipantInput.Type; + +export const LifecycleArchiveResult = Schema.Struct({ + archived: Schema.Boolean, + droppedExchangeIds: Schema.Array(ExchangeId), +}); +export type LifecycleArchiveResult = typeof LifecycleArchiveResult.Type; diff --git a/apps/server/src/j5/a2a/index.ts b/apps/server/src/j5/a2a/index.ts index 00133e96cef7..4899384c6300 100644 --- a/apps/server/src/j5/a2a/index.ts +++ b/apps/server/src/j5/a2a/index.ts @@ -1,4 +1,19 @@ export * from "./contracts.ts"; +export { + A2AArchiveFacts, + A2AArchiveFactsError, + A2AArchivePlacementFactsProviderError, + A2AArchivePlacementFactsProvider, + layer as archiveFactsLayer, + placementFactsUnavailableLayer, +} from "./ArchiveFactsService.ts"; +export type { + A2AArchiveFactsShape, + A2AArchivePlacementFactsProviderShape, + ArchivePlacementSubtree, + OpenExchangeArchiveFact, + ThreadPreArchiveFacts, +} from "./ArchiveFactsService.ts"; export * from "./decider.ts"; export { A2AHomeCommandConflictError, @@ -16,6 +31,16 @@ export type { RegisterAtCreationInput, } from "./HomeRegistrar.ts"; export * from "./LedgerService.ts"; +export { + A2ALifecycleCounterpartyStateError, + A2ALifecycleBridgeError, + A2ALifecycleHumanArchiveNotAllowedError, + A2ALifecycleParticipantHomeStateError, + A2ALifecycleParticipantNotFoundError, + A2ALifecycleService, + formatLifecycleNotice, +} from "./LifecycleService.ts"; +export type { A2ALifecycleError, A2ALifecycleServiceShape } from "./LifecycleService.ts"; export * from "./Migrations.ts"; export * from "./placementContracts.ts"; export * from "./placementProvenance.ts"; diff --git a/apps/server/src/j5/a2a/migrations/008_LifecycleClosure.ts b/apps/server/src/j5/a2a/migrations/008_LifecycleClosure.ts new file mode 100644 index 000000000000..a44717c1cdc7 --- /dev/null +++ b/apps/server/src/j5/a2a/migrations/008_LifecycleClosure.ts @@ -0,0 +1,329 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // The exchange projection becomes terminal at Dropped. Foreign-key checks + // are deferred while the table is replaced so A4's human-inbox projection + // keeps referencing the stable table name. A9 never derives or edits its + // fields; the opaque snapshot below only defeats the parent's cascade. + yield* sql`PRAGMA defer_foreign_keys = ON`; + const humanInboxTables = yield* sql<{ readonly name: string }>` + SELECT name + FROM sqlite_master + WHERE type = 'table' AND name = 'j5_a2a_human_inbox' + `; + const hasHumanInbox = humanInboxTables[0] !== undefined; + if (hasHumanInbox) { + // Dropping the parent invokes A4's ON DELETE CASCADE even when foreign-key + // validation is deferred. Snapshot every opaque projection column so the + // parent rebuild has no observable effect on A4-owned data. + yield* sql` + CREATE TEMP TABLE j5_a2a_human_inbox_lifecycle_snapshot AS + SELECT + person_id, + squadron_id, + exchange_id, + sender_id, + intent, + urgency, + latest_message_id, + latest_message, + opened_seq, + opened_at, + status, + terminal_seq, + terminal_at, + terminal_disposition, + terminal_cause, + terminal_facts, + terminal_notice_message_id + FROM j5_a2a_human_inbox + `; + } + yield* sql` + CREATE TABLE j5_a2a_exchange_lifecycle_new ( + squadron_id TEXT NOT NULL, + exchange_id TEXT NOT NULL, + sender_id TEXT NOT NULL, + receiver_id TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('open', 'closed', 'dropped')), + intent TEXT NOT NULL CHECK (length(trim(intent)) > 0), + urgency TEXT CHECK (urgency IN ('blocking', 'soon', 'fyi')), + opened_seq INTEGER NOT NULL, + closed_seq INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (squadron_id, exchange_id), + FOREIGN KEY (squadron_id) REFERENCES j5_a2a_squadron(id) ON DELETE CASCADE + ) + `; + yield* sql` + INSERT INTO j5_a2a_exchange_lifecycle_new ( + squadron_id, + exchange_id, + sender_id, + receiver_id, + status, + intent, + urgency, + opened_seq, + closed_seq, + created_at, + updated_at + ) + SELECT + squadron_id, + exchange_id, + sender_id, + receiver_id, + status, + intent, + urgency, + opened_seq, + closed_seq, + created_at, + updated_at + FROM j5_a2a_exchange + `; + yield* sql`DROP TABLE j5_a2a_exchange`; + yield* sql`ALTER TABLE j5_a2a_exchange_lifecycle_new RENAME TO j5_a2a_exchange`; + if (hasHumanInbox) { + yield* sql` + INSERT INTO j5_a2a_human_inbox ( + person_id, + squadron_id, + exchange_id, + sender_id, + intent, + urgency, + latest_message_id, + latest_message, + opened_seq, + opened_at, + status, + terminal_seq, + terminal_at, + terminal_disposition, + terminal_cause, + terminal_facts, + terminal_notice_message_id + ) + SELECT + person_id, + squadron_id, + exchange_id, + sender_id, + intent, + urgency, + latest_message_id, + latest_message, + opened_seq, + opened_at, + status, + terminal_seq, + terminal_at, + terminal_disposition, + terminal_cause, + terminal_facts, + terminal_notice_message_id + FROM j5_a2a_human_inbox_lifecycle_snapshot + `; + yield* sql`DROP TABLE j5_a2a_human_inbox_lifecycle_snapshot`; + } + yield* sql` + CREATE UNIQUE INDEX j5_a2a_exchange_open_pair_idx + ON j5_a2a_exchange(squadron_id, sender_id, receiver_id) + WHERE status = 'open' + `; + yield* sql` + CREATE INDEX j5_a2a_exchange_id_idx + ON j5_a2a_exchange(exchange_id) + `; + + yield* sql` + CREATE TABLE j5_a2a_delivery_lifecycle_new ( + squadron_id TEXT NOT NULL, + message_id TEXT NOT NULL, + command_id TEXT NOT NULL, + sent_seq INTEGER NOT NULL, + sender_id TEXT NOT NULL, + receiver_id TEXT NOT NULL, + receiver_squadron_id TEXT NOT NULL, + exchange_id TEXT, + exchange_role TEXT NOT NULL CHECK (exchange_role IN ( + 'none', + 'ask', + 'followup', + 'reply', + 'terminal_notice' + )), + correlation_id TEXT NOT NULL, + message_text TEXT NOT NULL CHECK (length(message_text) > 0), + status TEXT NOT NULL CHECK (status IN ( + 'pending', + 'retry_scheduled', + 'delivered', + 'alarmed' + )), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + last_error TEXT, + next_attempt_at TEXT, + delivered_seq INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + envelope_channel TEXT NOT NULL CHECK (envelope_channel IN ( + 'peer', + 'silence_notice', + 'lifecycle_notice' + )), + PRIMARY KEY (squadron_id, message_id), + FOREIGN KEY (squadron_id) REFERENCES j5_a2a_squadron(id) ON DELETE CASCADE + ) + `; + yield* sql` + INSERT INTO j5_a2a_delivery_lifecycle_new ( + squadron_id, + message_id, + command_id, + sent_seq, + sender_id, + receiver_id, + receiver_squadron_id, + exchange_id, + exchange_role, + correlation_id, + message_text, + status, + attempts, + last_error, + next_attempt_at, + delivered_seq, + created_at, + updated_at, + envelope_channel + ) + SELECT + squadron_id, + message_id, + command_id, + sent_seq, + sender_id, + receiver_id, + receiver_squadron_id, + exchange_id, + exchange_role, + correlation_id, + message_text, + status, + attempts, + last_error, + next_attempt_at, + delivered_seq, + created_at, + updated_at, + envelope_channel + FROM j5_a2a_delivery + `; + yield* sql`DROP TABLE j5_a2a_delivery`; + yield* sql`ALTER TABLE j5_a2a_delivery_lifecycle_new RENAME TO j5_a2a_delivery`; + yield* sql` + CREATE INDEX j5_a2a_delivery_drain_idx + ON j5_a2a_delivery(status, next_attempt_at, sent_seq) + `; + yield* sql` + CREATE INDEX j5_a2a_delivery_message_sender_idx + ON j5_a2a_delivery(message_id, sender_id) + `; + yield* sql` + CREATE UNIQUE INDEX j5_a2a_delivery_one_reply_idx + ON j5_a2a_delivery(exchange_id) + WHERE exchange_id IS NOT NULL AND exchange_role = 'reply' + `; + + yield* sql` + CREATE TABLE j5_a2a_comm_event_lifecycle_new ( + seq INTEGER NOT NULL, + squadron_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ( + 'exchange.opened', + 'message.sent', + 'message.received', + 'message.delivered', + 'message.delivery_failed', + 'exchange.closed', + 'exchange.dropped', + 'silence.notice', + 'participant.joined', + 'participant.left' + )), + sender TEXT, + receiver TEXT, + exchange_id TEXT, + correlation_id TEXT, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + command_id TEXT, + PRIMARY KEY (squadron_id, seq), + FOREIGN KEY (squadron_id) REFERENCES j5_a2a_squadron(id) ON DELETE RESTRICT, + CHECK (kind <> 'message.received' OR correlation_id IS NOT NULL) + ) + `; + yield* sql` + INSERT INTO j5_a2a_comm_event_lifecycle_new ( + seq, + squadron_id, + kind, + sender, + receiver, + exchange_id, + correlation_id, + payload, + created_at, + command_id + ) + SELECT + seq, + squadron_id, + kind, + sender, + receiver, + exchange_id, + correlation_id, + payload, + created_at, + command_id + FROM j5_a2a_comm_event + `; + yield* sql`DROP TABLE j5_a2a_comm_event`; + yield* sql`ALTER TABLE j5_a2a_comm_event_lifecycle_new RENAME TO j5_a2a_comm_event`; + yield* sql` + CREATE UNIQUE INDEX j5_a2a_comm_event_received_correlation_idx + ON j5_a2a_comm_event(squadron_id, correlation_id) + WHERE kind = 'message.received' + `; + yield* sql` + CREATE INDEX j5_a2a_comm_event_command_idx + ON j5_a2a_comm_event(command_id, squadron_id, seq) + WHERE command_id IS NOT NULL + `; + yield* sql` + CREATE UNIQUE INDEX j5_a2a_comm_event_agent_home_thread_idx + ON j5_a2a_comm_event(json_extract(payload, '$.participant.threadId')) + WHERE kind = 'participant.joined' + AND json_extract(payload, '$.participant.kind') = 'agent' + `; + + yield* sql` + CREATE TABLE j5_a2a_lifecycle_cursor ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + after_sequence INTEGER NOT NULL DEFAULT 0 CHECK (after_sequence >= 0), + updated_at TEXT + ) + `; + yield* sql` + INSERT INTO j5_a2a_lifecycle_cursor (singleton, after_sequence, updated_at) + VALUES (1, 0, NULL) + `; +}); diff --git a/apps/server/src/j5/a2a/runtimeLayer.test.ts b/apps/server/src/j5/a2a/runtimeLayer.test.ts index 2ce3905ee7fd..8b0038401c6b 100644 --- a/apps/server/src/j5/a2a/runtimeLayer.test.ts +++ b/apps/server/src/j5/a2a/runtimeLayer.test.ts @@ -9,6 +9,8 @@ import * as NodeSqliteClient from "../../persistence/NodeSqliteClient.ts"; import { runMigrations } from "../../persistence/Migrations.ts"; import { ThreadManagementService } from "../../orchestration-v2/ThreadManagementService.ts"; import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { A2AArchiveFacts } from "./ArchiveFactsService.ts"; +import { A2ALifecycleService } from "./LifecycleService.ts"; import { runJ5A2AMigrations } from "./Migrations.ts"; import { ParticipantPlacementService } from "./PlacementService.ts"; import { A2ASilenceDetector } from "./SilenceDetector.ts"; @@ -74,6 +76,8 @@ it.effect("shares one runtime across the combined HTTP and MCP-style route graph ParticipantPlacementService.pipe(Effect.asVoid), ); const silenceConsumer = Layer.effectDiscard(A2ASilenceDetector.pipe(Effect.asVoid)); + const lifecycleConsumer = Layer.effectDiscard(A2ALifecycleService.pipe(Effect.asVoid)); + const archiveFactsConsumer = Layer.effectDiscard(A2AArchiveFacts.pipe(Effect.asVoid)); const runtime = makeJ5A2ARuntimeLayer({ ledger: countedLedger }); yield* Layer.build( Layer.mergeAll( @@ -81,6 +85,8 @@ it.effect("shares one runtime across the combined HTTP and MCP-style route graph secondThreadConsumer, placementConsumer, silenceConsumer, + lifecycleConsumer, + archiveFactsConsumer, ).pipe( Layer.provideMerge(runtime), Layer.provide(countedThreadManagement), diff --git a/apps/server/src/j5/a2a/runtimeLayer.ts b/apps/server/src/j5/a2a/runtimeLayer.ts index afd47b9662c0..29e6aac691a7 100644 --- a/apps/server/src/j5/a2a/runtimeLayer.ts +++ b/apps/server/src/j5/a2a/runtimeLayer.ts @@ -1,11 +1,16 @@ import * as Layer from "effect/Layer"; +import { + layer as archiveFactsLayer, + placementFactsUnavailableLayer, +} from "./ArchiveFactsService.ts"; import { layer as deliveryWorkerLayer } from "./DeliveryWorker.ts"; import { live as deliveryTransportLayer } from "./DeliveryTransport.ts"; import { layer as homeRegistrarLayer } from "./HomeRegistrar.ts"; import { humanPersonRegistryLayer } from "./HumanPersonRegistry.ts"; import { layer as ledgerLayer } from "./LedgerService.ts"; import { layer as participantPlacementLayer } from "./PlacementService.ts"; +import { layer as lifecycleServiceLayer } from "./LifecycleService.ts"; import { layer as sendServiceLayer } from "./SendService.ts"; import { layer as silenceDetectorLayer } from "./SilenceDetector.ts"; import { layer as humanInboxLayer } from "./HumanInboxService.ts"; @@ -24,6 +29,12 @@ export const makeJ5A2ARuntimeLayer = ( const silenceDetectorProvided = silenceDetectorLayer.pipe( Layer.provideMerge(deliveryWorkerProvided), ); + const lifecycleServiceProvided = lifecycleServiceLayer.pipe( + Layer.provideMerge(deliveryWorkerProvided), + ); + const archiveFactsProvided = archiveFactsLayer.pipe( + Layer.provide(placementFactsUnavailableLayer), + ); return Layer.mergeAll( humanPersonRegistryLayer, homeRegistrarLayer, @@ -32,6 +43,8 @@ export const makeJ5A2ARuntimeLayer = ( silenceDetectorProvided, humanInboxLayer, participantPlacementLayer, + lifecycleServiceProvided, + archiveFactsProvided, ).pipe(Layer.provideMerge(ledgerProvided)); }; diff --git a/docs/j5/worklog/a2a-tickets/a3-silence-detector/index.md b/docs/j5/worklog/a2a-tickets/a3-silence-detector/index.md index c20e336ffaed..d0f6bfe37e72 100644 --- a/docs/j5/worklog/a2a-tickets/a3-silence-detector/index.md +++ b/docs/j5/worklog/a2a-tickets/a3-silence-detector/index.md @@ -18,6 +18,7 @@ When an agent that owes a reply goes quiet, the waiter learns _why_ — from pla - The five v1 states exactly as the plan's table: `turn-ended-no-reply` (with processed vs never-processed timestamp sub-split), `errored` (raw detail attached), `stopped/cancelled` (carries do-not-retry/do-not-replace instruction text), `awaiting-human` (human-knows vs human-doesn't-know via inbox delivery state), `blocked-on-peer` (**peer id stored structurally in the payload** — future cycle detection must need no ledger migration). - Notices append to the ledger (`silence.notice`) and deliver to the waiter through the A2 pipeline using the silence-notice envelope channel (clearly marked system signal, never styled as a peer message). - Notices inform; they NEVER auto-close exchanges. +- **A9 lifecycle cross-reference:** before committing a notice batch, A3 conditionally appends only if the Exchange is still open, with that guard inside the ledger's permit-protected transaction. A9 Dropped first means no silence row; A3 first means its true historical notice remains alongside the later lifecycle fact. ## Out of scope diff --git a/docs/j5/worklog/a2a-tickets/a9-lifecycle-closure.md b/docs/j5/worklog/a2a-tickets/a9-lifecycle-closure.md index 7f79235f234c..59786a22e123 100644 --- a/docs/j5/worklog/a2a-tickets/a9-lifecycle-closure.md +++ b/docs/j5/worklog/a2a-tickets/a9-lifecycle-closure.md @@ -1,32 +1,48 @@ --- -title: "A9 — Lifecycle closure: archive terminates obligations, loudly (R1/R2/R11)" +title: "A9 — Lifecycle closure: participant retirement terminates obligations, loudly (R1/R11)" kind: ticket -status: 0 +status: 1 --- # A9 — Lifecycle closure -**Governing artifacts:** design-review register R1, R2, R11 (../../product/design-review-2026-08-21.md). Staff against current `j5/main` head. **Hazard note (why this is blocking-tier): archives already happen in practice and currently strand open obligations silently.** +**Governing artifacts:** design-review register R1 and participant-side R11 (../../product/design-review-2026-08-21.md). Staff against current `j5/main` head. **Hazard note (why this is blocking-tier): participant retirement already happens in practice and currently strands open obligations silently.** ## Goal -Retiring a participant or a Squadron terminates its open obligations as a loud, evented act — waiters are told, nothing dangles, history stays readable. +Retiring a participant terminates its open obligations as a loud, evented act — waiters are told, nothing dangles, and history stays readable. ## Scope 1. **Participant archive termination (R1):** archiving a participant transitions its open Exchanges to **Dropped** with platform-authored terminal notices delivered to the affected counterparties through the normal pipeline — both dispositions: receiver-retired (waiter told: dropped, carries do-not-retry/do-not-replace) and sender-retired (the reply-owing party told the asker is gone; its obligation ends). Notices are fact bundles per R4. -2. **Squadron archive (R2):** a user-only Squadron archive operation (the Squadron entity has NO archive op today — this ticket adds it, user-only per E4 authority pattern) force-closing all open Exchanges via the R1 machinery, with visible inbox-row removal for affected humans. -3. **Active-view hiding (R11):** archived participants/Squadrons disappear from active views while their ledgers remain readable forever — no deletion, no truncation; projections gain the archived dimension. -4. **Write authority note:** this is NEW machinery, deliberately not an A3 extension — silence notices inform and never change Exchange state; archive termination is a state-transitioning write authority. Same pattern family (lifecycle fact → platform-authored ledger rows → pipeline delivery), separate service. +2. **Active-view hiding (R11, participant half):** `participant.left` removes the retired participant from active membership reads while its append-only ledger history remains readable forever. No extra participant archive column is needed because the real lifecycle writer already exercises this projection. +3. **Write authority note:** this is NEW machinery, deliberately not an A3 extension — silence notices inform and never change Exchange state; archive termination is a state-transitioning write authority. Same pattern family (lifecycle fact → platform-authored ledger rows → pipeline delivery), separate service. +4. **Committed thread bridge:** a bounded J5-owned reactor consumes committed `thread.archived` and `thread.deleted` events from a durable cursor and routes registered A2A homes through the same idempotent termination authority. Native threads without an A2A home are explicit no-ops. Restart, archive→delete, and replay must not duplicate terminal dispositions or notices. +5. **A3/A9 serialization rule:** A3 may derive a silence fact while an Exchange is open, but its ledger append must recheck `status = open` inside the same permit-protected transaction that commits the notice. If A9 has already committed Dropped, A3 emits nothing; if A3 commits first, both the silence fact and later Dropped fact remain true history. A3 remains informational and never transitions Exchange state. +6. **AR2 fact gathering only:** a read-only service reports a registered participant's open inbound and outbound Exchanges for the future human pre-archive dialog. Placement subtree data sits behind a provider and is structurally `unknown`, `none`, or `known`; the current provider returns `unknown` until the A6 component table is ratified. The service displays facts only and makes no archive/settle decision. + +## Scope amendment — Squadron archive deferred + +The settled archive-flow work ([product contract](../../product/features/archive-flow.md), [session ruling](../archive-flow-session-2026-08-29.md)) and Squadron SC4 ([feature definition](../../product/features/squadron.md)) supersede A9's original R2/Squadron half. All Squadron-archive-shaped work is deferred to SC4 polished-later: no operation, HTTP/MCP/server surface, warning/confirmation path, event, schema column, or projection dimension ships here. Migration `008_LifecycleClosure` contains no Squadron schema. Existing append-only ledgers already remain readable; a future SC4 implementation can compose the R1 participant termination machinery after its user contract ships. + +Known migration cost: rebuilding `j5_a2a_exchange` activates A4's `ON DELETE CASCADE`, so 008 snapshots and restores every A4 inbox column opaquely inside the one migration transaction. Any future rebuild of this parent faces the same hazard; recurring snapshot code must become a separately reviewed migration-framework helper rather than another copy-paste. + +AR2 surface sequencing: A4's sanctioned J5 HTTP routes layer now exists, but A9 supplies only the service-complete read boundary and defers route registration until the human archive-dialog surface is ratified. An MCP read tool was measured as technically possible and rejected because the consumer is the human pre-archive dialog, not an agent. Whether open Exchanges should affect upstream Settle remains an explicit future decision; this service only makes the facts demonstrable. + +Future composition note: A2S `archive_agent` will call `ArchiveFactsService` for the AR2 preflight read and call `LifecycleService.archiveParticipant` directly for the idempotent R1 termination. It must not duplicate either engine or archive-and-hope that the asynchronous reactor catches up. A9 intentionally adds no A2S handler or tool stub. ## Out of scope -Memos-on-archive warning (R35 — ships with the Memos feature, not here). Captain-invoked crew archive (R18–R22 — item-3 territory; this ticket's operations are user/platform-invoked only). Un-archive semantics (undesigned). +Squadron archive and AR1–AR4 warn/confirm behavior (SC4 polished-later), Memos-on-archive warning (R35 — ships with the Memos feature, not here), and Captain-invoked crew archive (R18–R22 — item-3 territory). A later upstream `thread.unarchived` event does not revive A2A participation, reopen Dropped exchanges, retract lifecycle notices, or recreate obligations. Re-entry remains undesigned and requires a separately ratified lifecycle policy. ## Dependencies -A2 merged (pipeline). Sequencing with registrar/A6 lane: none hard; coordinate migration numbering. +A2 and Registrar PR #10 are merged. A9 consumes Registrar's immutable `resolveThreadHome` history and complete-payload `participant.left` retirement. Fleet allocation is fixed: A4 `006_HumanNode`, then A6 `007_ParticipantPlacement`, then A9 `008_LifecycleClosure`. The final pre-PR rebase must verify that exact contiguous manifest; A9 must stop rather than self-adjust if it differs. ## Acceptance -Scripted scenarios: archive a participant owing a reply → waiter receives Dropped notice with do-not-retry facts, exchange terminal in ledger and projections; archive a participant who is owed a reply → counterparty notified, obligation ended; **archive an agent holding an open Exchange addressed to a PERSON (the person owes the reply) → the person's inbox row is removed/closed visibly and evented — never left dangling, never silently vanished (R1 sender-retired disposition on the inbox surface, per design-review authority's addition)**; Squadron archive → all open Exchanges closed via R1 path, human inbox rows removed visibly (evented), ledger fully readable post-archive; negative controls: a healthy participant's exchanges must be untouched by another's archive, and an archived participant must reject new sends with a state-naming error; baseline suite green. +Scripted scenarios: archive a participant owing a reply → waiter receives Dropped notice with do-not-retry facts, exchange terminal in ledger and projections; archive a participant who is owed a reply → counterparty notified, obligation ended; **archive an agent holding an open Exchange addressed to a PERSON (the person owes the reply) → the person's inbox row is removed/closed visibly and evented — never left dangling, never silently vanished (R1 sender-retired disposition on the inbox surface, per design-review authority's addition)**; committed deletion of a reply-owing participant yields the same single terminal disposition/notice; archive→delete and restart/replay never duplicate closure; deterministic A3-read→A9-drop→A3-append ordering emits no stale silence row, while A3-append→A9-drop preserves both rows; negative controls: same- and cross-Squadron healthy exchanges remain untouched, native no-home thread retirement is a no-op, no Squadron archive API/schema is introduced, and an A2A-retired participant rejects new sends with a state-naming error even after upstream unarchive; focused migration/runtime/domain suites green on the final post-A4/A6 rebase. + +Final proof order is binding: first make the deterministic seed harness green on the exact stationary post-rebase head; only after independent stationary review, run the R1 acceptance through a real Codex Luna participant in isolated disposable state, capture the resulting lifecycle evidence, and shut that state down. The standing test grant removes per-run approval for this final proof, but it does not permit an early live run or bypass Director usage-limit gates. + +The Luna run must use worktree-local `.t3/userdata`, never the live `~/.t3/userdata`; any realistic seed comes from a safe copy or `VACUUM INTO`, never a symlink. Capture the spawned server PID and tear down only that PID—no pattern kill or `pgrep`, because the worktree path is present in agent process arguments. Evidence is discriminating only when it captures the waiter's received lifecycle-notice text, the ledger's terminal Exchange state, and the persisted do-not-retry fact bundle; a clean run or absence of errors is not acceptance evidence.