diff --git a/FORK.md b/FORK.md index f6f0e4c6fdfa6..17891100fd68a 100644 --- a/FORK.md +++ b/FORK.md @@ -35,6 +35,17 @@ Treat these upstream areas as off-limits except for those explicit appended case - existing provider adapters and shared runtime modules - vendored references under `.repos` +### Sanctioned appended integration cases + +Against upstream pin `993407dd9e57f1edf2f5681d70140bfefeca93cc`, the complete A2A exception inventory is exactly these four cases. Line numbers identify this revision; the named symbol or test is the durable anchor after nearby upstream movement. + +1. A1's independent migration lane: `apps/server/src/persistence/Layers/Sqlite.ts:10` imports `runJ5A2AMigrations`, and `:42` runs it after upstream migrations. Introduced by `a064a87ac40ea2d2d936ba72008c95edeb8bbc2b` and merged in `521c50aa9bb6b4c7f55bc10a772822ec31129f2d`. +2. The one shared authenticated J5 MCP/runtime seam: `apps/server/src/mcp/McpHttpServer.ts:31` imports `J5McpIntegrationLive`, and `:247-248` append its sole entry to `layer`. Registration and runtime composition stay in `apps/server/src/j5/a2a/mcp/registration.ts`; A6 extends the J5-owned toolkit without another protected-file registration. +3. The internal delivery-dedup contract proof: `apps/server/src/orchestration-v2/runtimeLayer.test.ts:188-234`, test `replays an internal thread send without injecting a second message`. +4. The authenticated shared-toolkit integration proof: `apps/server/src/mcp/toolkits/worktree/registration.test.ts:15,70,83-90,128-131`, within test `production mcp layer lists worktree tools over http`. + +These are per-instance Director/Jackson-authorized exceptions, not standing category permission. The earlier fork rebrand is separately complete at `0c0de1acefea00a34f9529bb97be32ff5056cfcc`; its rebase-critical boundary is recorded in `BRANDING.md:1-5,24-35`. The supporting fork setup plan (`artifacts/fork-setup-plan/index.md:8-12`) and its six T1-T6 ticket artifacts are internal project records and are not present in this repository. + If a required change cannot fit this discipline, stop and review the exception before implementing it. ### Fork-owned migration lane diff --git a/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts b/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts new file mode 100644 index 0000000000000..ebc778738032b --- /dev/null +++ b/apps/server/src/j5/a2a/DeliveryTransport.integration.test.ts @@ -0,0 +1,491 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + MessageId, + type ModelSelection, + type OrchestrationV2ProviderSession, + type OrchestrationV2ProviderThread, + type OrchestrationV2TurnItem, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +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 CheckpointStore from "../../checkpointing/CheckpointStore.ts"; +import { ServerConfig } from "../../config.ts"; +import { layer as mcpSessionRegistryTestLayer } from "../../mcp/McpSessionRegistry.testkit.ts"; +import { OrchestrationEffectWorkerV2 } from "../../orchestration-v2/EffectWorker.ts"; +import { EventSinkV2 } from "../../orchestration-v2/EventSink.ts"; +import { OrchestratorV2 } from "../../orchestration-v2/Orchestrator.ts"; +import type { + ProviderAdapterV2Event, + ProviderAdapterV2Shape, + ProviderAdapterV2SteerInput, +} from "../../orchestration-v2/ProviderAdapter.ts"; +import { + OrchestrationV2EventSinkLayerLive, + OrchestrationV2LayerLive, +} from "../../orchestration-v2/runtimeLayer.ts"; +import { CodexProviderCapabilitiesV2 } from "../../orchestration-v2/Adapters/CodexAdapterV2.ts"; +import { + latestSteerableRun, + ThreadManagementService, + type ThreadManagementSendMode, +} from "../../orchestration-v2/ThreadManagementService.ts"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import type { ProviderInstance } from "../../provider/ProviderDriver.ts"; +import { ProviderInstanceRegistry } from "../../provider/Services/ProviderInstanceRegistry.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import * as VcsDriverRegistry from "../../vcs/VcsDriverRegistry.ts"; +import * as VcsProcess from "../../vcs/VcsProcess.ts"; +import { + A2ADeliveryTransport, + deliveryMessageId, + live as deliveryTransportLayer, +} from "./DeliveryTransport.ts"; +import { formatHumanEnvelope, formatPeerEnvelope } from "./EnvelopeFormatter.ts"; +import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { + CommCommandId, + EpicId, + ExchangeId, + GLOBAL_HUMAN_PARTICIPANT_ID, + LedgerMessageId, + ParticipantId, +} from "./contracts.ts"; + +const serverConfigLayer = ServerConfig.layerTest(process.cwd(), { + prefix: "t3-j5-a2a-delivery-transport-", +}); + +const modelSelection = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", +} satisfies ModelSelection; + +const vcsDriverRegistryTestLayer = VcsDriverRegistry.layer.pipe( + Layer.provide(VcsProcess.layer), + Layer.provide(serverConfigLayer), + Layer.provide(NodeServices.layer), +); + +const checkpointStoreTestLayer = CheckpointStore.layer.pipe( + Layer.provide(vcsDriverRegistryTestLayer), +); + +const driver = ProviderDriverKind.make("codex"); + +interface DeliveryInvocation { + readonly messageId: MessageId; + readonly mode: ThreadManagementSendMode; + readonly createdBy: "user" | "agent" | "system"; +} + +interface DeliveryHarness { + readonly deliveryInvocations: Ref.Ref>; + readonly steerInputs: Ref.Ref>; +} + +const makeOrchestrationAdapter = ( + steerInputs: Ref.Ref>, +): ProviderAdapterV2Shape => ({ + instanceId: modelSelection.instanceId, + driver, + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (sessionInput) => + Effect.gen(function* () { + const events = yield* PubSub.unbounded(); + const now = yield* DateTime.now; + const providerSession: OrchestrationV2ProviderSession = { + id: sessionInput.providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: sessionInput.runtimePolicy.cwd ?? process.cwd(), + model: sessionInput.modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; + + return { + instanceId: modelSelection.instanceId, + driver, + providerSessionId: sessionInput.providerSessionId, + providerSession, + events: Stream.fromPubSub(events), + ensureThread: (input) => + Effect.gen(function* () { + const createdAt = yield* DateTime.now; + return { + id: ProviderThreadId.make(`provider-thread:${input.threadId}`), + driver, + providerInstanceId: modelSelection.instanceId, + providerSessionId: sessionInput.providerSessionId, + appThreadId: input.threadId, + ownerNodeId: null, + nativeThreadRef: { + driver, + nativeId: `native-thread:${input.threadId}`, + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt, + updatedAt: createdAt, + } satisfies OrchestrationV2ProviderThread; + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (input) => + Effect.gen(function* () { + const startedAt = yield* DateTime.now; + yield* PubSub.publish(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${input.attemptId}`), + providerThreadId: input.providerThread.id, + nodeId: input.rootNodeId, + runAttemptId: input.attemptId, + nativeTurnRef: { + driver, + nativeId: `native-turn:${input.attemptId}`, + strength: "strong", + }, + ordinal: input.providerTurnOrdinal, + status: "running", + startedAt, + completedAt: null, + }, + }); + }), + steerTurn: (input) => Ref.update(steerInputs, (existing) => [...existing, input]), + interruptTurn: () => Effect.die("interruptTurn is unused by the A2 delivery seam test"), + respondToRuntimeRequest: () => + Effect.die("respondToRuntimeRequest is unused by the A2 delivery seam test"), + readThreadSnapshot: () => + Effect.die("readThreadSnapshot is unused by the A2 delivery seam test"), + rollbackThread: () => Effect.die("rollbackThread is unused by the A2 delivery seam test"), + forkThread: () => Effect.die("forkThread is unused by the A2 delivery seam test"), + }; + }), +}); + +const makeTestLayer = (harness: DeliveryHarness) => { + const orchestrationAdapter = makeOrchestrationAdapter(harness.steerInputs); + const providerInstance = { + instanceId: modelSelection.instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: "codex:j5-a2a-delivery-test", + }, + displayName: "Codex A2 delivery test", + enabled: true, + snapshot: {} as ProviderInstance["snapshot"], + orchestrationAdapter, + textGeneration: {} as ProviderInstance["textGeneration"], + } satisfies ProviderInstance; + const providerInstanceRegistryTestLayer = Layer.succeed(ProviderInstanceRegistry, { + getInstance: (instanceId) => + Effect.succeed(instanceId === providerInstance.instanceId ? providerInstance : undefined), + listInstances: Effect.succeed([providerInstance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.never, + }); + const orchestrationTestLayer = Layer.merge( + OrchestrationV2LayerLive, + OrchestrationV2EventSinkLayerLive, + ).pipe( + Layer.provide(mcpSessionRegistryTestLayer), + Layer.provide(checkpointStoreTestLayer), + Layer.provide(serverConfigLayer), + Layer.provide(ServerSettingsService.layerTest()), + Layer.provide(providerInstanceRegistryTestLayer), + Layer.provide(NodeServices.layer), + Layer.provideMerge(SqlitePersistenceMemory), + ); + const recordingThreadManagement = Layer.effect( + ThreadManagementService, + Effect.gen(function* () { + const threads = yield* ThreadManagementService; + return ThreadManagementService.of({ + ...threads, + sendToThread: (input) => + Ref.update(harness.deliveryInvocations, (existing) => [ + ...existing, + { messageId: input.messageId, mode: input.mode, createdBy: input.createdBy }, + ]).pipe(Effect.andThen(threads.sendToThread(input))), + }); + }), + ); + const recordedDeliveryTransport = deliveryTransportLayer.pipe( + Layer.provide(recordingThreadManagement), + ); + + return Layer.merge(ledgerLayer, recordedDeliveryTransport).pipe( + Layer.provideMerge(orchestrationTestLayer), + ); +}; + +const makeHarness = Effect.gen(function* () { + return { + deliveryInvocations: yield* Ref.make>([]), + steerInputs: yield* Ref.make>([]), + } satisfies DeliveryHarness; +}); + +const seedTarget = (suffix: string) => + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const ledger = yield* A2ALedger; + const threadId = ThreadId.make(`thread:j5-a2a-delivery-${suffix}`); + const projectId = ProjectId.make(`project:j5-a2a-delivery-${suffix}`); + const epicId = EpicId.make(`epic:j5-a2a-delivery-${suffix}`); + const senderId = ParticipantId.make(`agent:j5-a2a-delivery-${suffix}-sender`); + const receiverId = ParticipantId.make(`agent:j5-a2a-delivery-${suffix}-receiver`); + const exchangeId = ExchangeId.make(`exchange:j5-a2a-delivery-${suffix}`); + const messageId = LedgerMessageId.make(`message:j5-a2a-delivery-${suffix}`); + const createdAt = "2026-08-17T12:00:00.000Z"; + const message = `Reply through the real ${suffix} delivery seam.`; + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:j5-a2a-delivery-${suffix}-create-thread`), + threadId, + projectId, + title: `J5 A2A ${suffix} delivery target`, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: `/tmp/j5-a2a-delivery-${suffix}`, + }); + yield* ledger.createEpic({ + epic: { id: epicId, name: `J5 A2A ${suffix} delivery`, createdAt }, + }); + yield* ledger.appendEvents({ + commandId: CommCommandId.make(`command:j5-a2a-delivery-${suffix}-join-target`), + epicId, + acceptedAt: createdAt, + events: [ + { + kind: "participant.joined", + sender: null, + receiver: receiverId, + exchangeId: null, + correlationId: null, + payload: { + participant: { kind: "agent", id: receiverId, threadId }, + }, + createdAt, + }, + ], + }); + + return { + threadId, + projectId, + epicId, + senderId, + receiverId, + exchangeId, + messageId, + message, + delivery: { + originEpicId: epicId, + receiverEpicId: epicId, + messageId, + senderId, + receiverId, + exchangeId, + message, + }, + }; + }); + +it.effect("starts an idle recipient immediately without the implicit auto mode", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const threads = yield* ThreadManagementService; + const transport = yield* A2ADeliveryTransport; + const target = yield* seedTarget("idle"); + + yield* transport.deliverAgent(target.delivery); + yield* transport.deliverAgent(target.delivery); + + const projection = yield* threads.getThreadProjection(target.threadId); + const upstreamMessageId = deliveryMessageId(target.messageId); + const deliveredMessages = projection.messages.filter( + (candidate) => candidate.id === upstreamMessageId, + ); + assert.lengthOf(deliveredMessages, 1); + assert.equal( + deliveredMessages[0]?.text, + formatPeerEnvelope({ + senderId: target.senderId, + originEpicId: target.epicId, + exchangeId: target.exchangeId, + message: target.message, + }), + ); + assert.lengthOf(projection.runs, 1); + assert.equal( + projection.turnItems.find( + ( + candidate, + ): candidate is Extract => + candidate.type === "user_message" && candidate.messageId === upstreamMessageId, + )?.inputIntent, + "turn_start", + ); + assert.deepStrictEqual( + (yield* Ref.get(harness.deliveryInvocations)) + .filter((invocation) => invocation.messageId === upstreamMessageId) + .map((invocation) => invocation.mode), + ["queue", "queue"], + ); + }).pipe(Effect.provide(makeTestLayer(harness))); + }), +); + +it.effect("steers a busy recipient inside its active turn without queueing a later run", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const worker = yield* OrchestrationEffectWorkerV2; + const threads = yield* ThreadManagementService; + const transport = yield* A2ADeliveryTransport; + const target = yield* seedTarget("busy"); + const active = yield* threads.sendToThread({ + projectId: target.projectId, + commandId: CommandId.make("command:j5-a2a-delivery-busy-start"), + threadId: target.threadId, + messageId: MessageId.make("message:j5-a2a-delivery-busy-start"), + text: "Stay active until the peer message arrives.", + attachments: [], + mode: "auto", + createdBy: "user", + creationSource: "web", + }); + assert.equal(active.delivery, "started"); + + const runningEvent = yield* eventSink.stream({ threadId: target.threadId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.runHead, + Effect.forkChild({ startImmediately: true }), + ); + assert.isTrue(yield* worker.runOnce); + yield* Fiber.join(runningEvent); + const busyProjection = yield* threads.getThreadProjection(target.threadId); + assert.equal(latestSteerableRun(busyProjection)?.id, active.run.id); + + yield* transport.deliverAgent(target.delivery); + yield* transport.deliverAgent(target.delivery); + assert.isTrue(yield* worker.runOnce); + + const upstreamMessageId = deliveryMessageId(target.messageId); + const projection = yield* threads.getThreadProjection(target.threadId); + const deliveredMessages = projection.messages.filter( + (candidate) => candidate.id === upstreamMessageId, + ); + assert.lengthOf(deliveredMessages, 1); + assert.equal(deliveredMessages[0]?.runId, active.run.id); + assert.equal( + deliveredMessages[0]?.text, + formatPeerEnvelope({ + senderId: target.senderId, + originEpicId: target.epicId, + exchangeId: target.exchangeId, + message: target.message, + }), + ); + assert.lengthOf(projection.runs, 1); + assert.equal( + projection.turnItems.find( + ( + candidate, + ): candidate is Extract => + candidate.type === "user_message" && candidate.messageId === upstreamMessageId, + )?.inputIntent, + "steer", + ); + assert.deepStrictEqual( + (yield* Ref.get(harness.deliveryInvocations)) + .filter((invocation) => invocation.messageId === upstreamMessageId) + .map((invocation) => invocation.mode), + ["steer", "steer"], + ); + const steerInputs = yield* Ref.get(harness.steerInputs); + assert.lengthOf(steerInputs, 1); + assert.equal(steerInputs[0]?.runId, active.run.id); + assert.equal(steerInputs[0]?.message.text, deliveredMessages[0]?.text); + assert.isFalse(yield* worker.runOnce); + }).pipe(Effect.provide(makeTestLayer(harness))); + }), +); + +it.effect("attributes human-origin delivery to the user actor", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const threads = yield* ThreadManagementService; + const transport = yield* A2ADeliveryTransport; + const target = yield* seedTarget("human-origin"); + const humanMessageId = LedgerMessageId.make("message:j5-a2a-delivery-human-origin"); + const message = "Human-authored request delivered through A2A."; + + yield* transport.deliverAgent({ + ...target.delivery, + messageId: humanMessageId, + senderId: GLOBAL_HUMAN_PARTICIPANT_ID, + message, + }); + + const upstreamMessageId = deliveryMessageId(humanMessageId); + const projection = yield* threads.getThreadProjection(target.threadId); + const delivered = projection.messages.find((candidate) => candidate.id === upstreamMessageId); + assert.equal(delivered?.createdBy, "user"); + assert.equal( + delivered?.text, + formatHumanEnvelope({ + senderId: GLOBAL_HUMAN_PARTICIPANT_ID, + exchangeId: target.exchangeId, + message, + }), + ); + assert.deepStrictEqual( + (yield* Ref.get(harness.deliveryInvocations)).filter( + (invocation) => invocation.messageId === upstreamMessageId, + ), + [{ messageId: upstreamMessageId, mode: "queue", createdBy: "user" }], + ); + }).pipe(Effect.provide(makeTestLayer(harness))); + }), +); diff --git a/apps/server/src/j5/a2a/DeliveryTransport.ts b/apps/server/src/j5/a2a/DeliveryTransport.ts new file mode 100644 index 0000000000000..88f957b189044 --- /dev/null +++ b/apps/server/src/j5/a2a/DeliveryTransport.ts @@ -0,0 +1,181 @@ +import { CommandId, MessageId } 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 * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; +import { formatHumanEnvelope, formatPeerEnvelope } from "./EnvelopeFormatter.ts"; +import { + EpicId, + ExchangeId, + GLOBAL_HUMAN_PARTICIPANT_ID, + Participant, + ParticipantId, + type LedgerMessageId, +} from "./contracts.ts"; + +export class A2ADeliveryTargetError extends Schema.TaggedErrorClass()( + "A2ADeliveryTargetError", + { + participantId: Schema.String, + state: Schema.String, + }, +) { + override get message(): string { + return `Cannot deliver to ${this.participantId}: ${this.state}. Call list_participants before sending again.`; + } +} + +export class A2ADeliveryTransportError extends Schema.TaggedErrorClass()( + "A2ADeliveryTransportError", + { + operation: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export interface AgentDeliveryInput { + readonly originEpicId: EpicId; + readonly receiverEpicId: EpicId; + readonly messageId: LedgerMessageId; + readonly senderId: ParticipantId; + readonly receiverId: ParticipantId; + readonly exchangeId: ExchangeId | null; + readonly message: string; +} + +export interface HumanDeliveryInput extends AgentDeliveryInput { + readonly createdAt: string; +} + +export interface A2ADeliveryTransportShape { + readonly deliverAgent: ( + input: AgentDeliveryInput, + ) => Effect.Effect; + readonly deliverHuman: ( + input: HumanDeliveryInput, + ) => Effect.Effect; +} + +export class A2ADeliveryTransport extends Context.Service< + A2ADeliveryTransport, + A2ADeliveryTransportShape +>()("t3/j5/a2a/DeliveryTransport/A2ADeliveryTransport") {} + +const stablePart = (value: string) => encodeURIComponent(value); + +/** + * Every retry reuses this upstream command/message pair. A rejected upstream + * receipt therefore exhausts into the visible alarm instead of rotating an id + * that could double-inject after an ambiguous post-commit failure. + */ +export const deliveryCommandId = (messageId: LedgerMessageId) => + CommandId.make(`command:j5:a2a:delivery:${stablePart(messageId)}`); + +export const deliveryMessageId = (messageId: LedgerMessageId) => + MessageId.make(`message:j5:a2a:delivery:${stablePart(messageId)}`); + +interface MembershipRow { + readonly payload: string; +} + +const decodeParticipant = Schema.decodeUnknownEffect(Schema.fromJsonString(Participant)); + +export const live: Layer.Layer< + A2ADeliveryTransport, + never, + ThreadManagement.ThreadManagementService | SqlClient.SqlClient +> = Layer.effect( + A2ADeliveryTransport, + Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + const sql = yield* SqlClient.SqlClient; + + return A2ADeliveryTransport.of({ + deliverAgent: (input) => + Effect.gen(function* () { + const rows = yield* sql` + SELECT payload + FROM j5_a2a_epic_membership + WHERE epic_id = ${input.receiverEpicId} + AND participant_id = ${input.receiverId} + LIMIT 1 + `; + const row = rows[0]; + if (row === undefined) { + return yield* new A2ADeliveryTargetError({ + participantId: input.receiverId, + state: "membership disappeared before delivery", + }); + } + const participant = yield* decodeParticipant(row.payload); + if (participant.kind !== "agent") { + return yield* new A2ADeliveryTargetError({ + participantId: input.receiverId, + state: "participant is not an addressable agent thread", + }); + } + const target = yield* threads.getThreadProjection(participant.threadId); + // Interject peer traffic into a busy agent's active turn so blocked-on-peer + // work resumes before silence classification. Queue mode starts immediately + // when idle without reintroducing ThreadManagement's implicit auto branch. + const mode = + ThreadManagement.latestSteerableRun(target) === undefined ? "queue" : "steer"; + const envelope = + input.senderId === GLOBAL_HUMAN_PARTICIPANT_ID + ? formatHumanEnvelope({ + senderId: input.senderId, + exchangeId: input.exchangeId, + message: input.message, + }) + : formatPeerEnvelope({ + senderId: input.senderId, + originEpicId: input.originEpicId, + exchangeId: input.exchangeId, + message: input.message, + }); + yield* threads.sendToThread({ + projectId: target.thread.projectId, + commandId: deliveryCommandId(input.messageId), + threadId: participant.threadId, + messageId: deliveryMessageId(input.messageId), + text: envelope, + attachments: [], + mode, + createdBy: input.senderId === GLOBAL_HUMAN_PARTICIPANT_ID ? "user" : "agent", + creationSource: "mcp", + }); + }).pipe( + Effect.mapError( + (cause) => new A2ADeliveryTransportError({ operation: "deliver agent", cause }), + ), + ), + deliverHuman: (input) => + sql` + INSERT INTO j5_a2a_human_inbox_data ( + origin_epic_id, + message_id, + exchange_id, + sender_id, + payload, + created_at + ) VALUES ( + ${input.originEpicId}, + ${input.messageId}, + ${input.exchangeId}, + ${input.senderId}, + ${input.message}, + ${input.createdAt} + ) + ON CONFLICT(origin_epic_id, message_id) DO NOTHING + `.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 new file mode 100644 index 0000000000000..843572905d4c6 --- /dev/null +++ b/apps/server/src/j5/a2a/DeliveryWorker.test.ts @@ -0,0 +1,564 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ThreadId } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as NodeSqliteClient from "../../persistence/NodeSqliteClient.ts"; +import { ThreadManagementService } from "../../orchestration-v2/ThreadManagementService.ts"; +import { + A2ADeliveryHookError, + A2ADeliveryHooks, + A2ADeliveryWorker, + layerWithHooks as deliveryWorkerLayerWithHooks, +} from "./DeliveryWorker.ts"; +import { + A2ADeliveryTransport, + A2ADeliveryTransportError, + deliveryCommandId, + deliveryMessageId, + live as deliveryTransportLive, + type A2ADeliveryTransportShape, +} from "./DeliveryTransport.ts"; +import { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { runJ5A2AMigrations } from "./Migrations.ts"; +import { A2ASendService, layer as sendLayer } from "./SendService.ts"; +import { + CommCommandId, + EpicId, + GLOBAL_HUMAN_PARTICIPANT_ID, + ParticipantId, + type AgentParticipant, +} from "./contracts.ts"; + +const timestamp = "2026-08-16T12:00:00.000Z"; +const sender: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:delivery-sender"), + threadId: ThreadId.make("thread:delivery-sender"), +}; +const receiver: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:delivery-receiver"), + threadId: ThreadId.make("thread:delivery-receiver"), +}; + +const makeTestLayer = ( + transport: A2ADeliveryTransportShape, + hooks: A2ADeliveryHooks["Service"] = A2ADeliveryHooks.of({ + afterTransportSuccess: () => Effect.void, + }), +) => { + const database = NodeSqliteClient.layerMemory(); + const ledger = ledgerLayer.pipe(Layer.provide(database)); + const send = sendLayer.pipe(Layer.provide(ledger), Layer.provide(database)); + const transportLayer = Layer.succeed(A2ADeliveryTransport, A2ADeliveryTransport.of(transport)); + const worker = deliveryWorkerLayerWithHooks(false).pipe( + Layer.provide(ledger), + Layer.provide(database), + Layer.provide(transportLayer), + Layer.provide(Layer.succeed(A2ADeliveryHooks, hooks)), + ); + return Layer.mergeAll(database, ledger, send, transportLayer, worker); +}; + +const join = Effect.fn("test.j5.a2a.delivery.join")(function* ( + epicId: EpicId, + participant: AgentParticipant, + index: string, +) { + yield* (yield* A2ALedger).append({ + commandId: CommCommandId.make(`command:delivery:join:${index}`), + epicId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: participant.id, + exchangeId: null, + correlationId: null, + payload: { participant }, + createdAt: timestamp, + }, + }); +}); + +const seedSend = Effect.fn("test.j5.a2a.delivery.seedSend")(function* (crossEpic: boolean) { + yield* runJ5A2AMigrations(); + const ledgerService = yield* A2ALedger; + const senderEpicId = EpicId.make("epic:delivery:sender"); + const receiverEpicId = crossEpic ? EpicId.make("epic:delivery:receiver") : senderEpicId; + yield* ledgerService.createEpic({ + epic: { id: senderEpicId, name: "Delivery sender", createdAt: timestamp }, + }); + if (crossEpic) { + yield* ledgerService.createEpic({ + epic: { id: receiverEpicId, name: "Delivery receiver", createdAt: timestamp }, + }); + } + yield* join(senderEpicId, sender, "sender"); + yield* join(receiverEpicId, receiver, "receiver"); + const sent = yield* (yield* A2ASendService).send({ + commandId: CommCommandId.make( + crossEpic ? "command:delivery:cross-epic" : "command:delivery:same-epic", + ), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Delivery crash-window probe", + acceptedAt: timestamp, + }); + return { senderEpicId, receiverEpicId, sent }; +}); + +const crashWindowScenario = (poisonIds: boolean, crossEpic: boolean) => + Effect.gen(function* () { + const calls = yield* Ref.make(0); + const injections = yield* Ref.make(0); + const seen = yield* Ref.make(new Set()); + const hookCalls = yield* Ref.make(0); + const transport: A2ADeliveryTransportShape = { + deliverAgent: (input) => + Effect.gen(function* () { + const call = yield* Ref.updateAndGet(calls, (count) => count + 1); + const stableKey = `${deliveryCommandId(input.messageId)}:${deliveryMessageId(input.messageId)}`; + const key = poisonIds ? `${stableKey}:attempt:${call}` : stableKey; + const inserted = yield* Ref.modify(seen, (keys) => { + if (keys.has(key)) return [false, keys] as const; + const next = new Set(keys); + next.add(key); + return [true, next] as const; + }); + if (inserted) yield* Ref.update(injections, (count) => count + 1); + }).pipe( + Effect.mapError( + (cause) => new A2ADeliveryTransportError({ operation: "test delivery", cause }), + ), + ), + deliverHuman: () => Effect.void, + }; + const hooks = A2ADeliveryHooks.of({ + afterTransportSuccess: () => + Ref.updateAndGet(hookCalls, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 + ? Effect.fail( + new A2ADeliveryHookError({ + cause: "simulated crash after injection before delivery receipt", + }), + ) + : Effect.void, + ), + ), + }); + + const result = yield* Effect.gen(function* () { + const { receiverEpicId, sent } = yield* seedSend(crossEpic); + const worker = yield* A2ADeliveryWorker; + const sql = yield* SqlClient.SqlClient; + if (crossEpic) { + const before = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${receiverEpicId} AND kind = 'message.received' + `; + assert.equal(before[0]?.count, 0, "sender commit precedes receiver double-entry"); + const senderState = yield* sql<{ readonly status: string }>` + SELECT status + FROM j5_a2a_delivery + WHERE message_id = ${sent.messageId} + `; + assert.deepStrictEqual(senderState, [{ status: "pending" }]); + } + + const first = yield* worker.runOnce; + assert.equal(first?.state, "retry_scheduled"); + assert.equal(first?.attempt, 1); + yield* TestClock.adjust("250 millis"); + const replay = yield* worker.runOnce; + assert.equal(replay?.state, "delivered"); + assert.equal(replay?.attempt, 2); + + const received = crossEpic + ? yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${receiverEpicId} + AND kind = 'message.received' + AND correlation_id IS NOT NULL + ` + : [{ count: 0 }]; + const senderState = yield* sql<{ readonly status: string }>` + SELECT status + FROM j5_a2a_delivery + WHERE message_id = ${sent.messageId} + `; + return { + messageId: sent.messageId, + injectionCount: yield* Ref.get(injections), + receivedCount: received[0]?.count ?? 0, + senderState: senderState[0]?.status, + }; + }).pipe(Effect.provide(makeTestLayer(transport, hooks))); + return result; + }); + +it.effect("recovers the injected-but-unrecorded window with one upstream injection", () => + Effect.gen(function* () { + const result = yield* crashWindowScenario(false, false); + assert.equal(result.injectionCount, 1); + }), +); + +it.effect("negative control: poisoned retry ids are detected as a double injection", () => + Effect.gen(function* () { + const poisoned = yield* crashWindowScenario(true, false); + assert.equal(poisoned.injectionCount, 2); + assert.notEqual(poisoned.injectionCount, 1); + }), +); + +it.effect("serializes manual runOnce calls against a concurrent drain", () => + Effect.gen(function* () { + const calls = yield* Ref.make(0); + const firstEntered = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + const transport: A2ADeliveryTransportShape = { + deliverAgent: () => + Ref.updateAndGet(calls, (count) => count + 1).pipe( + Effect.flatMap((call) => + call === 1 + ? Deferred.succeed(firstEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirst)), + ) + : Effect.void, + ), + ), + deliverHuman: () => Effect.void, + }; + + yield* Effect.gen(function* () { + yield* seedSend(false); + const worker = yield* A2ADeliveryWorker; + const runOnceFiber = yield* worker.runOnce.pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(firstEntered); + const drainFiber = yield* worker.drain.pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.yieldNow; + assert.equal(yield* Ref.get(calls), 1, "drain cannot claim the in-flight delivery"); + + yield* Deferred.succeed(releaseFirst, undefined); + assert.equal((yield* Fiber.join(runOnceFiber))?.state, "delivered"); + assert.deepStrictEqual(yield* Fiber.join(drainFiber), []); + assert.equal(yield* Ref.get(calls), 1); + }).pipe(Effect.provide(makeTestLayer(transport))); + }), +); + +it.effect("cross-epic half-write recovery records exactly one receiver entry", () => + Effect.gen(function* () { + const result = yield* crashWindowScenario(false, true); + assert.equal(result.injectionCount, 1); + assert.equal(result.receivedCount, 1); + assert.equal(result.senderState, "delivered"); + }), +); + +it.effect("closes a cross-epic exchange through the paired reply entry", () => + Effect.gen(function* () { + const injections = yield* Ref.make(0); + const transport: A2ADeliveryTransportShape = { + deliverAgent: () => Ref.update(injections, (count) => count + 1), + deliverHuman: () => Effect.void, + }; + yield* Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const ledgerService = yield* A2ALedger; + const sendService = yield* A2ASendService; + const worker = yield* A2ADeliveryWorker; + const sql = yield* SqlClient.SqlClient; + const senderEpicId = EpicId.make("epic:exchange:sender"); + const receiverEpicId = EpicId.make("epic:exchange:receiver"); + yield* ledgerService.createEpic({ + epic: { id: senderEpicId, name: "Exchange sender", createdAt: timestamp }, + }); + yield* ledgerService.createEpic({ + epic: { id: receiverEpicId, name: "Exchange receiver", createdAt: timestamp }, + }); + yield* join(senderEpicId, sender, "exchange-sender"); + yield* join(receiverEpicId, receiver, "exchange-receiver"); + + const opened = yield* sendService.send({ + commandId: CommCommandId.make("command:exchange:cross:open"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Please reply across epics.", + expectReply: true, + intent: "Prove cross-epic reply closure", + acceptedAt: timestamp, + }); + assert.equal(opened.exchangeState, "open"); + assert.equal((yield* worker.runOnce)?.state, "delivered"); + + const replyInput = { + commandId: CommCommandId.make("command:exchange:cross:reply"), + senderThreadId: receiver.threadId, + to: sender.id, + message: "Cross-epic reply delivered.", + exchangeId: opened.exchangeId!, + acceptedAt: timestamp, + } as const; + const reply = yield* sendService.send(replyInput); + assert.equal(reply.exchangeState, "closing"); + + const duplicateReply = yield* Effect.flip( + sendService.send({ + ...replyInput, + commandId: CommCommandId.make("command:exchange:cross:duplicate-reply"), + }), + ); + assert.equal(duplicateReply._tag, "A2AExchangeAlreadyAnsweredError"); + assert.equal((yield* worker.runOnce)?.state, "delivered"); + + const replay = yield* sendService.send(replyInput); + assert.deepStrictEqual(replay, reply, "command replay preserves its pre-delivery result"); + const exchange = yield* sql<{ readonly status: string }>` + SELECT status + FROM j5_a2a_exchange + WHERE exchange_id = ${opened.exchangeId!} + `; + assert.deepStrictEqual(exchange, [{ status: "closed" }]); + const paired = yield* sql<{ readonly epic_id: string; readonly count: number }>` + SELECT epic_id, COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE kind = 'message.received' + AND epic_id IN (${senderEpicId}, ${receiverEpicId}) + GROUP BY epic_id + ORDER BY epic_id + `; + assert.deepStrictEqual( + paired, + [ + { epic_id: receiverEpicId, count: 1 }, + { epic_id: senderEpicId, count: 1 }, + ].sort((left, right) => left.epic_id.localeCompare(right.epic_id)), + ); + const pairedPayloads = yield* sql<{ + readonly epic_id: string; + readonly payload: string; + }>` + SELECT epic_id, payload + FROM j5_a2a_comm_event + WHERE kind = 'message.received' + AND epic_id IN (${senderEpicId}, ${receiverEpicId}) + ORDER BY epic_id + `; + assert.deepStrictEqual( + pairedPayloads.map((row) => ({ + epicId: row.epic_id, + exchangeRole: (JSON.parse(row.payload) as { message: { exchangeRole: string } }).message + .exchangeRole, + })), + [ + { epicId: receiverEpicId, exchangeRole: "ask" }, + { epicId: senderEpicId, exchangeRole: "reply" }, + ].sort((left, right) => left.epicId.localeCompare(right.epicId)), + ); + assert.equal(yield* Ref.get(injections), 2); + }).pipe(Effect.provide(makeTestLayer(transport))); + }), +); + +it.effect("startup reconciliation drains a persisted cross-epic half-write after restart", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "j5-a2a-restart-" }); + const filename = path.join(directory, "state.sqlite"); + const firstDatabase = NodeSqliteClient.layer({ filename }); + const firstLedger = ledgerLayer.pipe(Layer.provide(firstDatabase)); + const firstSend = sendLayer.pipe(Layer.provide(firstLedger), Layer.provide(firstDatabase)); + const firstLayer = Layer.mergeAll(firstDatabase, firstLedger, firstSend); + const persisted = yield* Effect.scoped( + Effect.gen(function* () { + const seeded = yield* seedSend(true); + const sql = yield* SqlClient.SqlClient; + const received = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${seeded.receiverEpicId} AND kind = 'message.received' + `; + assert.equal(received[0]?.count, 0); + return seeded; + }).pipe(Effect.provide(firstLayer)), + ); + + const injectionCount = yield* Ref.make(0); + const transportEntered = yield* Deferred.make(); + const releaseTransport = yield* Deferred.make(); + const transport: A2ADeliveryTransportShape = { + deliverAgent: () => Ref.update(injectionCount, (count) => count + 1), + deliverHuman: () => Effect.void, + }; + const secondDatabase = NodeSqliteClient.layer({ filename }); + const secondLedger = ledgerLayer.pipe(Layer.provide(secondDatabase)); + const secondTransport = Layer.succeed( + A2ADeliveryTransport, + A2ADeliveryTransport.of(transport), + ); + const secondWorker = deliveryWorkerLayerWithHooks(true).pipe( + Layer.provide(secondLedger), + Layer.provide(secondDatabase), + Layer.provide(secondTransport), + Layer.provide( + Layer.succeed( + A2ADeliveryHooks, + A2ADeliveryHooks.of({ + afterTransportSuccess: () => + Deferred.succeed(transportEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseTransport)), + ), + }), + ), + ), + ); + const secondLayer = Layer.mergeAll( + secondDatabase, + secondLedger, + secondTransport, + secondWorker, + ); + yield* Effect.scoped( + Effect.gen(function* () { + const worker = yield* A2ADeliveryWorker; + yield* Deferred.await(transportEntered); + const milestoneStream = yield* worker.subscribeMilestones; + const milestoneFiber = yield* milestoneStream.pipe( + Stream.runHead, + Effect.forkChild({ startImmediately: true }), + ); + yield* Deferred.succeed(releaseTransport, undefined); + const reconciled = yield* Fiber.join(milestoneFiber).pipe(Effect.map(Option.getOrThrow)); + assert.equal(reconciled?.state, "delivered"); + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${persisted.receiverEpicId} + AND kind = 'message.received' + `; + assert.equal(rows[0]?.count, 1); + assert.equal(yield* Ref.get(injectionCount), 1); + }).pipe(Effect.provide(secondLayer)), + ); + }), + ).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("forces repeated delivery failure into a visible alarm", () => + Effect.gen(function* () { + const failure = new A2ADeliveryTransportError({ + operation: "forced failure", + cause: "recipient unavailable", + }); + const transport: A2ADeliveryTransportShape = { + deliverAgent: () => Effect.fail(failure), + deliverHuman: () => Effect.fail(failure), + }; + yield* Effect.gen(function* () { + yield* seedSend(false); + const worker = yield* A2ADeliveryWorker; + assert.equal((yield* worker.runOnce)?.state, "retry_scheduled"); + yield* TestClock.adjust("250 millis"); + assert.equal((yield* worker.runOnce)?.state, "retry_scheduled"); + yield* TestClock.adjust("500 millis"); + const alarm = yield* worker.runOnce; + assert.equal(alarm?.state, "alarmed"); + assert.equal(alarm?.attempt, 3); + const alarms = yield* worker.listAlarms; + assert.lengthOf(alarms, 1); + assert.equal(alarms[0]?.messageId, alarm?.messageId); + assert.include(alarms[0]?.lastError ?? "", "recipient unavailable"); + }).pipe(Effect.provide(makeTestLayer(transport))); + }), +); + +it.effect("delivers to the human through the idempotent inbox-data transport", () => + 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 threadManagement = Layer.mock(ThreadManagementService)({}); + const transport = deliveryTransportLive.pipe( + Layer.provide(database), + Layer.provide(threadManagement), + ); + 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 epicId = EpicId.make("epic:delivery:human"); + yield* ledgerService.createEpic({ + epic: { id: epicId, name: "Human delivery", createdAt: timestamp }, + }); + yield* join(epicId, sender, "human-sender"); + yield* ledgerService.append({ + commandId: CommCommandId.make("command:delivery:join:human"), + epicId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: GLOBAL_HUMAN_PARTICIPANT_ID, + exchangeId: null, + correlationId: null, + payload: { participant: { kind: "human" } }, + createdAt: timestamp, + }, + }); + const sent = yield* (yield* A2ASendService).send({ + commandId: CommCommandId.make("command:delivery:human"), + senderThreadId: sender.threadId, + to: GLOBAL_HUMAN_PARTICIPANT_ID, + message: "Human inbox payload", + acceptedAt: timestamp, + }); + const delivery = yield* (yield* A2ADeliveryWorker).runOnce; + assert.equal(delivery?.state, "delivered"); + + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ + readonly message_id: string; + readonly payload: string; + }>` + SELECT message_id, payload + FROM j5_a2a_human_inbox_data + WHERE origin_epic_id = ${epicId} + `; + assert.deepStrictEqual(rows, [ + { message_id: sent.messageId, payload: "Human inbox payload" }, + ]); + assert.isNull(yield* (yield* A2ADeliveryWorker).runOnce); + }).pipe(Effect.provide(layer)); + }), +); diff --git a/apps/server/src/j5/a2a/DeliveryWorker.ts b/apps/server/src/j5/a2a/DeliveryWorker.ts new file mode 100644 index 0000000000000..b9ee5910f96c3 --- /dev/null +++ b/apps/server/src/j5/a2a/DeliveryWorker.ts @@ -0,0 +1,422 @@ +import * as Cause from "effect/Cause"; +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 PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; +import * as Semaphore from "effect/Semaphore"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import type { SqlError } from "effect/unstable/sql/SqlError"; +import type * as Scope from "effect/Scope"; + +import config from "./delivery-config.v1.json" with { type: "json" }; +import { A2ADeliveryTransport, type A2ADeliveryTransportError } from "./DeliveryTransport.ts"; +import { + CommCommandId, + type CommEvent, + CorrelationId, + EpicId, + ExchangeId, + GLOBAL_HUMAN_PARTICIPANT_ID, + LedgerMessageId, + ParticipantId, + type DeliveryAlarm, + type DeliveryMilestone, +} from "./contracts.ts"; +import { A2ALedger, type A2ALedgerError } from "./LedgerService.ts"; + +export const A2A_DELIVERY_CONFIG_VERSION = config.version; + +interface DeliveryRow { + readonly epic_id: string; + readonly message_id: string; + readonly sent_seq: number; + readonly sender_id: string; + readonly receiver_id: string; + readonly receiver_epic_id: string; + readonly exchange_id: string | null; + readonly exchange_role: "none" | "ask" | "followup" | "reply"; + readonly correlation_id: string; + readonly message_text: string; + readonly status: "pending" | "retry_scheduled" | "delivered" | "alarmed"; + readonly attempts: number; + readonly created_at: string; +} + +interface OpenExchangeRow { + readonly epic_id: string; + readonly exchange_id: string; + readonly sender_id: string; + readonly receiver_id: string; +} + +export interface DeliveryAttempt { + readonly epicId: EpicId; + readonly messageId: LedgerMessageId; + readonly attempt: number; +} + +export interface A2ADeliveryHooksShape { + readonly afterTransportSuccess: ( + attempt: DeliveryAttempt, + ) => Effect.Effect; +} + +export class A2ADeliveryHookError extends Schema.TaggedErrorClass()( + "A2ADeliveryHookError", + { cause: Schema.Defect() }, +) {} + +export class A2ADeliveryWorkerError extends Schema.TaggedErrorClass()( + "A2ADeliveryWorkerError", + { operation: Schema.String, cause: Schema.Defect() }, +) {} + +export class A2ADeliveryHooks extends Context.Service()( + "t3/j5/a2a/DeliveryWorker/A2ADeliveryHooks", +) {} + +export const noopHooks = Layer.succeed( + A2ADeliveryHooks, + A2ADeliveryHooks.of({ afterTransportSuccess: () => Effect.void }), +); + +export interface A2ADeliveryWorkerShape { + readonly notify: Effect.Effect; + readonly runOnce: Effect.Effect; + readonly drain: Effect.Effect, A2ADeliveryWorkerError>; + readonly listAlarms: Effect.Effect, A2ADeliveryWorkerError>; + readonly subscribeMilestones: Effect.Effect, never, Scope.Scope>; +} + +export class A2ADeliveryWorker extends Context.Service()( + "t3/j5/a2a/DeliveryWorker/A2ADeliveryWorker", +) {} + +type A2ADeliveryAttemptError = + | A2ALedgerError + | A2ADeliveryTransportError + | A2ADeliveryHookError + | SqlError; + +const errorText = (cause: Cause.Cause) => + Cause.pretty(cause).slice(0, 4_000); + +const workerError = (operation: string) => (cause: unknown) => + new A2ADeliveryWorkerError({ operation, cause }); + +const backoffMs = (attempt: number) => + Math.min(config.initialBackoffMs * 2 ** Math.max(0, attempt - 1), config.maximumBackoffMs); + +const commandId = (operation: string, messageId: LedgerMessageId, attempt?: number) => + CommCommandId.make( + ["command", "j5", "a2a", operation, encodeURIComponent(messageId), attempt] + .filter((part) => part !== undefined) + .join(":"), + ); + +const makeLayer = (daemon: boolean) => + Layer.effect( + A2ADeliveryWorker, + Effect.gen(function* () { + const ledger = yield* A2ALedger; + const transport = yield* A2ADeliveryTransport; + const hooks = yield* A2ADeliveryHooks; + const sql = yield* SqlClient.SqlClient; + const wakeups = yield* Queue.unbounded(); + const milestones = yield* PubSub.unbounded(); + const drainPermit = yield* Semaphore.make(1); + + const appendReceiverEntry = Effect.fn("j5.a2a.delivery.appendReceiverEntry")(function* ( + row: DeliveryRow, + ) { + if (row.epic_id === row.receiver_epic_id) return; + const originEpicId = EpicId.make(row.epic_id); + const receiverEpicId = EpicId.make(row.receiver_epic_id); + const exchangeId = row.exchange_id === null ? null : ExchangeId.make(row.exchange_id); + const receivedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + const events: Array = [ + { + kind: "message.received", + sender: ParticipantId.make(row.sender_id), + receiver: ParticipantId.make(row.receiver_id), + exchangeId, + correlationId: CorrelationId.make(row.correlation_id), + payload: { + originEpicId, + message: { + messageId: row.message_id, + text: row.message_text, + originEpicId, + receiverEpicId, + exchangeRole: row.exchange_role, + }, + }, + createdAt: receivedAt, + }, + ]; + if (exchangeId !== null) { + const exchanges = yield* sql` + SELECT epic_id, exchange_id, sender_id, receiver_id + FROM j5_a2a_exchange + WHERE epic_id = ${receiverEpicId} + AND exchange_id = ${exchangeId} + AND status = 'open' + AND sender_id = ${row.receiver_id} + AND receiver_id = ${row.sender_id} + LIMIT 1 + `; + if (exchanges[0] !== undefined) { + events.push({ + kind: "exchange.closed", + sender: ParticipantId.make(row.sender_id), + receiver: ParticipantId.make(row.receiver_id), + exchangeId, + correlationId: CorrelationId.make(row.correlation_id), + payload: { replyMessageId: LedgerMessageId.make(row.message_id) }, + createdAt: receivedAt, + }); + } + } + yield* ledger.appendEvents({ + commandId: commandId("receive", LedgerMessageId.make(row.message_id)), + epicId: receiverEpicId, + acceptedAt: receivedAt, + events, + }); + }); + + const attemptDelivery = Effect.fn("j5.a2a.delivery.attempt")(function* ( + row: DeliveryRow, + attempt: number, + ) { + const originEpicId = EpicId.make(row.epic_id); + const receiverEpicId = EpicId.make(row.receiver_epic_id); + const messageId = LedgerMessageId.make(row.message_id); + const senderId = ParticipantId.make(row.sender_id); + const receiverId = ParticipantId.make(row.receiver_id); + const exchangeId = row.exchange_id === null ? null : ExchangeId.make(row.exchange_id); + yield* appendReceiverEntry(row); + if (receiverId === GLOBAL_HUMAN_PARTICIPANT_ID) { + yield* transport.deliverHuman({ + originEpicId, + receiverEpicId, + messageId, + senderId, + receiverId, + exchangeId, + message: row.message_text, + createdAt: row.created_at, + }); + } else { + yield* transport.deliverAgent({ + originEpicId, + receiverEpicId, + messageId, + senderId, + receiverId, + exchangeId, + message: row.message_text, + }); + } + yield* hooks.afterTransportSuccess({ epicId: originEpicId, messageId, attempt }); + const deliveredAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + yield* ledger.appendEvents({ + commandId: commandId("delivered", messageId), + epicId: originEpicId, + acceptedAt: deliveredAt, + events: [ + { + kind: "message.delivered", + sender: senderId, + receiver: receiverId, + exchangeId, + correlationId: CorrelationId.make(row.correlation_id), + payload: { + messageId, + attempt, + channel: receiverId === GLOBAL_HUMAN_PARTICIPANT_ID ? "human" : "agent", + }, + createdAt: deliveredAt, + }, + ], + }); + }); + + const recordFailure = Effect.fn("j5.a2a.delivery.recordFailure")(function* ( + row: DeliveryRow, + attempt: number, + cause: Cause.Cause, + ) { + const failedAtDate = yield* DateTime.now; + const failedAt = DateTime.formatIso(failedAtDate); + const alarmed = attempt >= config.alarmAfterAttempts; + const nextAttemptAt = alarmed + ? null + : DateTime.formatIso(DateTime.add(failedAtDate, { milliseconds: backoffMs(attempt) })); + const messageId = LedgerMessageId.make(row.message_id); + yield* ledger.appendEvents({ + commandId: commandId("failed", messageId, attempt), + epicId: EpicId.make(row.epic_id), + acceptedAt: failedAt, + events: [ + { + kind: "message.delivery_failed", + sender: ParticipantId.make(row.sender_id), + receiver: ParticipantId.make(row.receiver_id), + exchangeId: row.exchange_id === null ? null : ExchangeId.make(row.exchange_id), + correlationId: CorrelationId.make(row.correlation_id), + payload: { + messageId, + attempt, + error: errorText(cause), + nextAttemptAt, + alarmed, + }, + createdAt: failedAt, + }, + ], + }); + return { + epicId: EpicId.make(row.epic_id), + messageId, + state: alarmed ? ("alarmed" as const) : ("retry_scheduled" as const), + attempt, + } satisfies DeliveryMilestone; + }); + + const runOnceRaw = Effect.gen(function* () { + const now = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + const rows = yield* sql` + SELECT + epic_id, + message_id, + sent_seq, + sender_id, + receiver_id, + receiver_epic_id, + exchange_id, + exchange_role, + correlation_id, + message_text, + status, + attempts, + created_at + FROM j5_a2a_delivery + WHERE status IN ('pending', 'retry_scheduled') + AND (next_attempt_at IS NULL OR next_attempt_at <= ${now}) + ORDER BY sent_seq, epic_id, message_id + LIMIT 1 + `; + const row = rows[0]; + if (row === undefined) return null; + const attempt = row.attempts + 1; + const exit = yield* Effect.exit(attemptDelivery(row, attempt)); + const milestone = + exit._tag === "Success" + ? ({ + epicId: EpicId.make(row.epic_id), + messageId: LedgerMessageId.make(row.message_id), + state: "delivered", + attempt, + } satisfies DeliveryMilestone) + : yield* recordFailure(row, attempt, exit.cause); + yield* PubSub.publish(milestones, milestone); + return milestone; + }); + const runOnceEffect = runOnceRaw.pipe(Effect.mapError(workerError("run one delivery"))); + const runOnce: A2ADeliveryWorkerShape["runOnce"] = drainPermit.withPermit(runOnceEffect); + + const drainEffect = Effect.fn("j5.a2a.delivery.drain")(function* () { + const completed: Array = []; + while (true) { + const milestone = yield* runOnceEffect; + if (milestone === null) return completed; + completed.push(milestone); + } + }); + const drain: A2ADeliveryWorkerShape["drain"] = drainPermit.withPermit(drainEffect()); + + const nextDelay = Effect.fn("j5.a2a.delivery.nextDelay")(function* () { + const rows = yield* sql<{ readonly next_attempt_at: string | null }>` + SELECT next_attempt_at + FROM j5_a2a_delivery + WHERE status IN ('pending', 'retry_scheduled') + ORDER BY next_attempt_at IS NOT NULL, next_attempt_at, sent_seq + LIMIT 1 + `; + const value = rows[0]?.next_attempt_at; + if (value === undefined) return null; + if (value === null) return 0; + const now = yield* DateTime.now; + return Math.max(0, Date.parse(value) - DateTime.toEpochMillis(now)); + }); + + const runDaemon = Effect.forever( + Effect.gen(function* () { + yield* drain; + const delay = yield* nextDelay(); + if (delay === null) { + yield* Queue.take(wakeups); + } else if (delay === 0) { + yield* Effect.yieldNow; + } else { + yield* Effect.raceFirst(Queue.take(wakeups), Effect.sleep(Duration.millis(delay))); + } + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("J5 A2A delivery drain failed", { cause }).pipe( + Effect.andThen( + Effect.raceFirst( + Queue.take(wakeups), + Effect.sleep(Duration.millis(config.initialBackoffMs)), + ), + ), + ), + ), + ), + ); + if (daemon) { + yield* Effect.forkScoped(runDaemon); + } + yield* Queue.offer(wakeups, undefined); + + return A2ADeliveryWorker.of({ + notify: Queue.offer(wakeups, undefined).pipe(Effect.asVoid), + runOnce, + drain, + listAlarms: sql<{ + readonly epic_id: string; + readonly message_id: string; + readonly attempts: number; + readonly last_error: string; + }>` + SELECT epic_id, message_id, attempts, last_error + FROM j5_a2a_delivery + WHERE status = 'alarmed' + ORDER BY updated_at, epic_id, message_id + `.pipe( + Effect.map((rows) => + rows.map((row) => ({ + epicId: EpicId.make(row.epic_id), + messageId: LedgerMessageId.make(row.message_id), + attempts: row.attempts, + lastError: row.last_error, + })), + ), + Effect.mapError(workerError("list delivery alarms")), + ), + subscribeMilestones: PubSub.subscribe(milestones).pipe( + Effect.map((subscription) => Stream.fromSubscription(subscription)), + ), + }); + }), + ); + +export const manualLayer = makeLayer(false).pipe(Layer.provide(noopHooks)); +export const layer = makeLayer(true).pipe(Layer.provide(noopHooks)); +export const layerWithHooks = (daemon: boolean) => makeLayer(daemon); diff --git a/apps/server/src/j5/a2a/EnvelopeFormatter.test.ts b/apps/server/src/j5/a2a/EnvelopeFormatter.test.ts new file mode 100644 index 0000000000000..bce699fa52ef2 --- /dev/null +++ b/apps/server/src/j5/a2a/EnvelopeFormatter.test.ts @@ -0,0 +1,79 @@ +import { assert, it } from "@effect/vitest"; + +import { + A2A_ENVELOPE_VERSION, + A2A_LIST_TOOL_DESCRIPTION, + A2A_SEND_TOOL_DESCRIPTION, + formatHumanEnvelope, + formatPeerEnvelope, + formatSilenceNoticeEnvelope, +} from "./EnvelopeFormatter.ts"; +import { EpicId, ExchangeId, ParticipantId } from "./contracts.ts"; + +it("renders the versioned peer envelope with exact reply semantics", () => { + const rendered = formatPeerEnvelope({ + senderId: ParticipantId.make("agent:sender"), + originEpicId: EpicId.make("epic:origin"), + exchangeId: ExchangeId.make("exchange:one"), + message: "Please verify the worker.", + }); + + assert.equal(A2A_ENVELOPE_VERSION, 5); + assert.include(rendered, "Cross-agent message"); + assert.notMatch(rendered, /\b(?:J5|A2A)\b/); + assert.include(rendered, "agent:sender"); + assert.include(rendered, "epic:origin"); + assert.include(rendered, "Please verify the worker."); + assert.include(rendered, 'send_message(to="agent:sender", exchange_id="exchange:one"'); + assert.include(rendered, "Reply once"); + assert.notInclude(rendered, "{{"); +}); + +it("labels platform-authored silence without internal product branding", () => { + const rendered = formatSilenceNoticeEnvelope({ + noticeType: "peer unavailable", + message: "No reply was delivered.", + }); + + assert.include(rendered, "Cross-agent messaging system notice: peer unavailable"); + assert.include(rendered, "platform-authored delivery signal"); + assert.notMatch(rendered, /\b(?:J5|A2A)\b/); +}); + +it("does not interpret caller text as an envelope template", () => { + const message = "Preserve this literal token: {{exchangeInstruction}}"; + const rendered = formatPeerEnvelope({ + senderId: ParticipantId.make("agent:sender"), + originEpicId: EpicId.make("epic:origin"), + exchangeId: ExchangeId.make("exchange:one"), + message, + }); + + assert.include(rendered, message); + assert.equal(rendered.match(/send_message\(/g)?.length, 1); +}); + +it("tells agents that human-origin exchanges require an explicit tool reply", () => { + const rendered = formatHumanEnvelope({ + senderId: ParticipantId.make("human:global"), + exchangeId: ExchangeId.make("exchange:human"), + message: "Please report status.", + }); + + assert.include(rendered, "The human is not watching this chat"); + assert.include(rendered, 'exchange_id="exchange:human"'); + assert.include(A2A_SEND_TOOL_DESCRIPTION, "returns after the sender ledger commit"); + assert.include(A2A_SEND_TOOL_DESCRIPTION, "Durably send one message"); + assert.include(A2A_LIST_TOOL_DESCRIPTION, "List reachable message recipients"); + for (const description of [A2A_SEND_TOOL_DESCRIPTION, A2A_LIST_TOOL_DESCRIPTION]) { + assert.include(description, "native thread without a registered home epic"); + assert.include(description, "wrapper-spawned agent"); + assert.include(description, "controlled test seeding"); + assert.include(description, "home-epic registrar + A6 creation integrations follow-up"); + assert.notMatch(description, /ask the user|product workflow|list_participants again/i); + } + assert.notMatch( + [rendered, A2A_SEND_TOOL_DESCRIPTION, A2A_LIST_TOOL_DESCRIPTION].join("\n"), + /\b(?:J5|A2A)\b/, + ); +}); diff --git a/apps/server/src/j5/a2a/EnvelopeFormatter.ts b/apps/server/src/j5/a2a/EnvelopeFormatter.ts new file mode 100644 index 0000000000000..079f0f0762598 --- /dev/null +++ b/apps/server/src/j5/a2a/EnvelopeFormatter.ts @@ -0,0 +1,54 @@ +import config from "./envelopes.v1.json" with { type: "json" }; + +import type { EpicId, ExchangeId, ParticipantId } from "./contracts.ts"; + +export const A2A_ENVELOPE_VERSION = config.version; +export const A2A_SEND_TOOL_DESCRIPTION = config.sendToolDescription; +export const A2A_LIST_TOOL_DESCRIPTION = config.listToolDescription; + +const render = (template: string, values: Readonly>): string => + template.replace(/\{\{([^{}]+)\}\}/g, (placeholder, name: string) => values[name] ?? placeholder); + +const deliveryInstruction = (input: { + readonly senderId: ParticipantId; + readonly exchangeId: ExchangeId | null; +}) => + input.exchangeId === null + ? config.oneShotInstruction + : render(config.replyInstruction, { + senderId: input.senderId, + exchangeId: input.exchangeId, + }); + +export const formatPeerEnvelope = (input: { + readonly senderId: ParticipantId; + readonly originEpicId: EpicId; + readonly exchangeId: ExchangeId | null; + readonly message: string; +}): string => + render(config.peerMessage, { + senderId: input.senderId, + originEpicId: input.originEpicId, + message: input.message, + exchangeInstruction: deliveryInstruction(input), + }); + +export const formatHumanEnvelope = (input: { + readonly senderId: ParticipantId; + readonly exchangeId: ExchangeId | null; + readonly message: string; +}): string => + render(config.humanMessage, { + message: input.message, + exchangeInstruction: deliveryInstruction(input), + }); + +/** A3 supplies notice derivation; A2 owns this channel's stable rendering shape. */ +export const formatSilenceNoticeEnvelope = (input: { + readonly noticeType: string; + readonly message: string; +}): string => + render(config.silenceNotice, { + noticeType: input.noticeType, + message: input.message, + }); diff --git a/apps/server/src/j5/a2a/LedgerService.test.ts b/apps/server/src/j5/a2a/LedgerService.test.ts index d1141fd47c823..8afa37c17d269 100644 --- a/apps/server/src/j5/a2a/LedgerService.test.ts +++ b/apps/server/src/j5/a2a/LedgerService.test.ts @@ -24,6 +24,7 @@ import { CorrelationId, Epic, EpicId, + LedgerMessageId, ParticipantId, type AppendCommEventCommand, type CommEvent, @@ -42,7 +43,7 @@ const fileLedgerLayer = (filename: string) => ledgerLayer.pipe(Layer.provideMerge(NodeSqliteClient.layer({ filename }))); const messageEvent = (index: number): CommEvent => ({ - kind: "message.sent", + kind: "silence.notice", sender: ParticipantId.make("agent:sender"), receiver: ParticipantId.make("agent:receiver"), exchangeId: null, @@ -51,6 +52,53 @@ const messageEvent = (index: number): CommEvent => ({ createdAt: timestamp, }); +it.effect("routes single-event append through command ids and A2 projections", () => + Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const ledger = yield* A2ALedger; + const sql = yield* SqlClient.SqlClient; + const epicId = EpicId.make("epic:single-append-projection"); + const commandId = CommCommandId.make("command:single-append-projection"); + const messageId = LedgerMessageId.make("message:single-append-projection"); + yield* ledger.createEpic({ + epic: { id: epicId, name: "Single append projection", createdAt: timestamp }, + }); + yield* ledger.append({ + commandId, + epicId, + acceptedAt: timestamp, + event: { + kind: "message.sent", + sender: ParticipantId.make("agent:single:sender"), + receiver: ParticipantId.make("agent:single:receiver"), + exchangeId: null, + correlationId: CorrelationId.make("correlation:single-append-projection"), + payload: { + messageId, + text: "Single append remains deliverable.", + originEpicId: epicId, + receiverEpicId: epicId, + exchangeRole: "none", + }, + createdAt: timestamp, + }, + }); + + const rows = yield* sql<{ + readonly command_id: string; + readonly message_id: string; + readonly status: string; + }>` + SELECT command_id, message_id, status + FROM j5_a2a_delivery + WHERE message_id = ${messageId} + `; + assert.deepStrictEqual(rows, [ + { command_id: commandId, message_id: messageId, status: "pending" }, + ]); + }).pipe(Effect.provide(memoryLedgerLayer())), +); + const appendCommand = ( epicId: EpicId, index: number, @@ -364,7 +412,15 @@ it.effect("enforces one message.received correlation per receiver epic", () => createdAt: timestamp, }; yield* ledger.append(appendCommand(epicId, 1, receivedEvent)); - const error = yield* Effect.flip(ledger.append(appendCommand(epicId, 2, receivedEvent))); + const failedCommand = CommCommandId.make(`command:${epicId}:2`); + const error = yield* Effect.flip( + ledger.appendEvents({ + commandId: failedCommand, + epicId, + acceptedAt: timestamp, + events: [receivedEvent], + }), + ); assert.isTrue(isA2AStorageError(error)); const rows = yield* sql<{ readonly count: number }>` SELECT COUNT(*) AS count @@ -372,5 +428,102 @@ it.effect("enforces one message.received correlation per receiver epic", () => WHERE epic_id = ${epicId} AND kind = 'message.received' `; assert.equal(rows[0]?.count, 1); + + const receipts = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_command_receipt + WHERE command_id = ${failedCommand} + `; + assert.equal(receipts[0]?.count, 0, "the failed event insert rolls back its receipt"); + + const retried = yield* ledger.appendEvents({ + commandId: failedCommand, + epicId, + acceptedAt: timestamp, + events: [ + { + ...receivedEvent, + correlationId: CorrelationId.make("correlation:retry-after-rollback"), + }, + ], + }); + assert.isTrue(retried.committed, "the rolled-back command id remains reusable"); + }).pipe(Effect.provide(memoryLedgerLayer())), +); + +it.effect("rejects delivery transitions without a projected message row", () => + Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const ledger = yield* A2ALedger; + const sql = yield* SqlClient.SqlClient; + const epicId = EpicId.make("epic:missing-delivery-projection"); + yield* ledger.createEpic({ + epic: { id: epicId, name: "Missing delivery projection", createdAt: timestamp }, + }); + const transitions: ReadonlyArray<{ readonly name: string; readonly event: CommEvent }> = [ + { + name: "delivered", + event: { + kind: "message.delivered", + sender: ParticipantId.make("agent:delivery:sender"), + receiver: ParticipantId.make("agent:delivery:receiver"), + exchangeId: null, + correlationId: CorrelationId.make("correlation:missing-delivered"), + payload: { + messageId: LedgerMessageId.make("message:missing-delivered"), + attempt: 1, + channel: "agent", + }, + createdAt: timestamp, + }, + }, + { + name: "delivery-failed", + event: { + kind: "message.delivery_failed", + sender: ParticipantId.make("agent:delivery:sender"), + receiver: ParticipantId.make("agent:delivery:receiver"), + exchangeId: null, + correlationId: CorrelationId.make("correlation:missing-delivery-failed"), + payload: { + messageId: LedgerMessageId.make("message:missing-delivery-failed"), + attempt: 1, + error: "forced missing projection", + nextAttemptAt: timestamp, + alarmed: false, + }, + createdAt: timestamp, + }, + }, + ]; + + for (const [index, transition] of transitions.entries()) { + const commandId = CommCommandId.make(`command:missing-delivery:${transition.name}`); + const error = yield* Effect.flip( + ledger.appendEvents({ + commandId, + epicId, + acceptedAt: timestamp, + events: [transition.event], + }), + ); + assert.isTrue(isA2AStorageError(error)); + + const retried = yield* ledger.appendEvents({ + commandId, + epicId, + acceptedAt: timestamp, + events: [messageEvent(index + 100)], + }); + assert.isTrue(retried.committed, "the failed projection rolls back its command receipt"); + } + + const transitionRows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${epicId} + AND kind IN ('message.delivered', 'message.delivery_failed') + `; + assert.equal(transitionRows[0]?.count, 0); }).pipe(Effect.provide(memoryLedgerLayer())), ); diff --git a/apps/server/src/j5/a2a/LedgerService.ts b/apps/server/src/j5/a2a/LedgerService.ts index 5f683afd297b4..4b7a242b67a83 100644 --- a/apps/server/src/j5/a2a/LedgerService.ts +++ b/apps/server/src/j5/a2a/LedgerService.ts @@ -9,11 +9,17 @@ import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { + type AppendCommEventsCommand, CommCommandReceipt, type AppendCommEventCommand, type CommEventPage, type CreateEpicCommand, Epic, + ExchangeClosedPayload, + ExchangeOpenedPayload, + MessageDeliveredPayload, + MessageDeliveryFailedPayload, + MessageSentPayload, type EpicId, type LedgerCursor, Membership, @@ -79,11 +85,20 @@ export interface AppendResult { readonly committed: boolean; } +export interface AppendEventsResult { + readonly receipt: CommCommandReceipt; + readonly events: ReadonlyArray; + readonly committed: boolean; +} + export interface A2ALedgerShape { readonly createEpic: (command: CreateEpicCommand) => Effect.Effect; readonly listEpics: () => Effect.Effect, A2ALedgerError>; readonly readEpic: (epicId: EpicId) => Effect.Effect; readonly append: (command: AppendCommEventCommand) => Effect.Effect; + readonly appendEvents: ( + command: AppendCommEventsCommand, + ) => Effect.Effect; readonly readEvents: (input: { readonly epicId: EpicId; readonly cursor: LedgerCursor; @@ -139,6 +154,11 @@ const decodeEpic = Schema.decodeUnknownEffect(Epic); const decodeStoredEvent = Schema.decodeUnknownEffect(StoredCommEvent); const decodeReceipt = Schema.decodeUnknownEffect(CommCommandReceipt); const decodeMembership = Schema.decodeUnknownEffect(Membership); +const decodeExchangeOpened = Schema.decodeUnknownEffect(ExchangeOpenedPayload); +const decodeExchangeClosed = Schema.decodeUnknownEffect(ExchangeClosedPayload); +const decodeMessageSent = Schema.decodeUnknownEffect(MessageSentPayload); +const decodeMessageDelivered = Schema.decodeUnknownEffect(MessageDeliveredPayload); +const decodeMessageDeliveryFailed = Schema.decodeUnknownEffect(MessageDeliveryFailedPayload); const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Json)); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Json)); @@ -239,6 +259,155 @@ export const layer: Layer.Layer = Layer.e `; }); + const applyA2Projection = Effect.fn("j5.a2a.applyA2Projection")(function* ( + event: StoredCommEvent, + commandId: string, + ) { + switch (event.kind) { + case "exchange.opened": { + const payload = yield* decodeExchangeOpened(event.payload); + if (event.sender === null || event.receiver === null || event.exchangeId === null) { + return yield* new A2AStorageError({ operation: "project opened exchange" }); + } + yield* sql` + INSERT INTO j5_a2a_exchange ( + epic_id, + exchange_id, + sender_id, + receiver_id, + status, + intent, + urgency, + opened_seq, + closed_seq, + created_at, + updated_at + ) VALUES ( + ${event.epicId}, + ${event.exchangeId}, + ${event.sender}, + ${event.receiver}, + 'open', + ${payload.intent}, + ${payload.urgency}, + ${event.seq}, + NULL, + ${event.createdAt}, + ${event.createdAt} + ) + `; + return; + } + case "exchange.closed": { + yield* decodeExchangeClosed(event.payload); + if (event.exchangeId === null) { + return yield* new A2AStorageError({ operation: "project closed exchange" }); + } + yield* sql` + UPDATE j5_a2a_exchange + SET + status = 'closed', + closed_seq = ${event.seq}, + updated_at = ${event.createdAt} + WHERE epic_id = ${event.epicId} + 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) { + return yield* new A2AStorageError({ operation: "project sent message" }); + } + yield* sql` + INSERT INTO j5_a2a_delivery ( + epic_id, + message_id, + command_id, + sent_seq, + sender_id, + receiver_id, + receiver_epic_id, + exchange_id, + exchange_role, + correlation_id, + message_text, + status, + attempts, + last_error, + next_attempt_at, + delivered_seq, + created_at, + updated_at + ) VALUES ( + ${event.epicId}, + ${payload.messageId}, + ${commandId}, + ${event.seq}, + ${event.sender}, + ${event.receiver}, + ${payload.receiverEpicId}, + ${event.exchangeId}, + ${payload.exchangeRole}, + ${event.correlationId}, + ${payload.text}, + 'pending', + 0, + NULL, + NULL, + NULL, + ${event.createdAt}, + ${event.createdAt} + ) + `; + return; + } + case "message.delivered": { + const payload = yield* decodeMessageDelivered(event.payload); + const rows = yield* sql<{ readonly message_id: string }>` + UPDATE j5_a2a_delivery + SET + status = 'delivered', + attempts = ${payload.attempt}, + last_error = NULL, + next_attempt_at = NULL, + delivered_seq = ${event.seq}, + updated_at = ${event.createdAt} + WHERE epic_id = ${event.epicId} AND message_id = ${payload.messageId} + RETURNING message_id + `; + if (rows[0] === undefined) { + return yield* new A2AStorageError({ operation: "project delivered message" }); + } + return; + } + case "message.delivery_failed": { + const payload = yield* decodeMessageDeliveryFailed(event.payload); + const rows = yield* sql<{ readonly message_id: string }>` + UPDATE j5_a2a_delivery + SET + status = ${payload.alarmed ? "alarmed" : "retry_scheduled"}, + attempts = ${payload.attempt}, + last_error = ${payload.error}, + next_attempt_at = ${payload.nextAttemptAt}, + updated_at = ${event.createdAt} + WHERE epic_id = ${event.epicId} AND message_id = ${payload.messageId} + RETURNING message_id + `; + if (rows[0] === undefined) { + return yield* new A2AStorageError({ operation: "project failed message delivery" }); + } + return; + } + case "message.received": + case "silence.notice": + case "participant.joined": + case "participant.left": + return; + } + }); + const listMembershipEffect = Effect.fn("j5.a2a.listMembership")(function* (epicId: EpicId) { yield* ensureEpic(epicId); const rows = yield* sql` @@ -250,20 +419,22 @@ export const layer: Layer.Layer = Layer.e return yield* Effect.forEach(rows, membershipFromRow, { concurrency: 1 }); }); - const appendEffect = Effect.fn("j5.a2a.append")(function* (command: AppendCommEventCommand) { + const appendEventsEffect = Effect.fn("j5.a2a.appendEvents")(function* ( + command: AppendCommEventsCommand, + ) { const result = yield* sql.withTransaction( Effect.gen(function* () { yield* ensureEpic(command.epicId); - const pending = decideAppendCommEvent(command)[0]; const sequenceRows = yield* sql<{ readonly next_seq: number }>` SELECT COALESCE(MAX(seq), 0) + 1 AS next_seq FROM j5_a2a_comm_event WHERE epic_id = ${command.epicId} `; - const seq = sequenceRows[0]?.next_seq; - if (seq === undefined) { - return yield* new A2AStorageError({ operation: "allocate communication sequence" }); + 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, @@ -276,7 +447,7 @@ export const layer: Layer.Layer = Layer.e ${command.epicId}, 'comm.append', ${command.acceptedAt}, - ${seq} + ${resultSeq} ) ON CONFLICT(command_id) DO NOTHING RETURNING command_id @@ -291,9 +462,7 @@ export const layer: Layer.Layer = Layer.e `; const row = receiptRows[0]; if (row === undefined) { - return yield* new A2AStorageError({ - operation: "read replayed command receipt", - }); + return yield* new A2AStorageError({ operation: "read replayed batch receipt" }); } if (row.epic_id !== command.epicId) { return yield* new CommCommandConflictError({ @@ -314,60 +483,76 @@ export const layer: Layer.Layer = Layer.e payload, created_at FROM j5_a2a_comm_event - WHERE epic_id = ${command.epicId} AND seq = ${row.result_seq} - LIMIT 1 + WHERE epic_id = ${command.epicId} AND command_id = ${command.commandId} + ORDER BY seq `; - const eventRow = eventRows[0]; - if (eventRow === undefined) { - return yield* new A2AStorageError({ - operation: "read replayed command event", - }); + if (eventRows.length === 0) { + return yield* new A2AStorageError({ operation: "read replayed batch events" }); } return { receipt: yield* receiptFromRow(row), - event: yield* eventFromRow(eventRow), + events: yield* Effect.forEach(eventRows, eventFromRow, { concurrency: 1 }), committed: false as const, }; } - const payload = yield* encodeJson(pending.payload); - yield* sql` - INSERT INTO j5_a2a_comm_event ( - seq, - epic_id, - kind, - sender, - receiver, - exchange_id, - correlation_id, - payload, - created_at - ) VALUES ( - ${seq}, - ${pending.epicId}, - ${pending.kind}, - ${pending.sender}, - ${pending.receiver}, - ${pending.exchangeId}, - ${pending.correlationId}, - ${payload}, - ${pending.createdAt} - ) - `; - const event = yield* decodeStoredEvent({ seq, ...pending }); - yield* applyMembership(event); - const receipt = yield* decodeReceipt({ - commandId: command.commandId, - epicId: command.epicId, - commandType: "comm.append", - acceptedAt: command.acceptedAt, - resultSeq: seq, - }); - return { receipt, event, committed: true as const }; + const events: Array = []; + for (const [index, candidate] of command.events.entries()) { + const pending = decideAppendCommEvent({ + commandId: command.commandId, + epicId: command.epicId, + 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, + epic_id, + kind, + sender, + receiver, + exchange_id, + correlation_id, + payload, + created_at, + command_id + ) VALUES ( + ${seq}, + ${pending.epicId}, + ${pending.kind}, + ${pending.sender}, + ${pending.receiver}, + ${pending.exchangeId}, + ${pending.correlationId}, + ${payload}, + ${pending.createdAt}, + ${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, + epicId: command.epicId, + commandType: "comm.append", + acceptedAt: command.acceptedAt, + resultSeq, + }), + events, + committed: true as const, + }; }), ); if (result.committed) { - yield* PubSub.publish(committed, result.event); + for (const event of result.events) { + yield* PubSub.publish(committed, event); + } } return result; }); @@ -378,8 +563,16 @@ export const layer: Layer.Layer = Layer.e yield* sql` INSERT INTO j5_a2a_epic (id, name, created_at) VALUES (${command.epic.id}, ${command.epic.name}, ${command.epic.createdAt}) + ON CONFLICT(id) DO NOTHING `; - return command.epic; + return yield* epicFromRow( + (yield* sql` + SELECT id, name, created_at + FROM j5_a2a_epic + WHERE id = ${command.epic.id} + LIMIT 1 + `)[0]!, + ); }).pipe(Effect.mapError(preserveDomainError("create epic"))), listEpics: () => Effect.gen(function* () { @@ -399,8 +592,30 @@ export const layer: Layer.Layer = Layer.e }).pipe(Effect.mapError(preserveDomainError("read epic"))), append: (command) => appendPermit - .withPermit(appendEffect(command)) + .withPermit( + appendEventsEffect({ + commandId: command.commandId, + epicId: command.epicId, + acceptedAt: command.acceptedAt, + events: [command.event], + }).pipe( + Effect.flatMap((result) => { + const event = result.events[0]; + return event === undefined + ? Effect.fail(new A2AStorageError({ operation: "read single appended event" })) + : Effect.succeed({ + receipt: result.receipt, + event, + committed: result.committed, + }); + }), + ), + ) .pipe(Effect.mapError(preserveDomainError("append communication event"))), + appendEvents: (command) => + appendPermit + .withPermit(appendEventsEffect(command)) + .pipe(Effect.mapError(preserveDomainError("append communication events"))), readEvents: ({ epicId, cursor, limit }) => Effect.gen(function* () { yield* ensureEpic(epicId); diff --git a/apps/server/src/j5/a2a/Migrations.test.ts b/apps/server/src/j5/a2a/Migrations.test.ts index 31c423c7c4d14..e0a128a9a536f 100644 --- a/apps/server/src/j5/a2a/Migrations.test.ts +++ b/apps/server/src/j5/a2a/Migrations.test.ts @@ -29,10 +29,13 @@ it.effect("tracks J5 A2A migrations independently from upstream migrations", () `; assert.equal(upstream[0]?.migration_id, upstreamMigrationManifest.at(-1)?.[0]); - assert.deepStrictEqual(j5, [{ migration_id: 1, name: "EpicCommunicationLedger" }]); + assert.deepStrictEqual(j5, [ + { migration_id: 1, name: "EpicCommunicationLedger" }, + { migration_id: 2, name: "SendDeliverReply" }, + ]); assert.deepStrictEqual( migrationEntries.map(([id]) => id), - [1], + [1, 2], ); }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), ); @@ -49,7 +52,10 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c 'j5_a2a_epic', 'j5_a2a_comm_event', 'j5_a2a_comm_command_receipt', - 'j5_a2a_epic_membership' + 'j5_a2a_epic_membership', + 'j5_a2a_exchange', + 'j5_a2a_delivery', + 'j5_a2a_human_inbox_data' ) ORDER BY name `; @@ -59,7 +65,13 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c WHERE type = 'index' AND name IN ( 'j5_a2a_comm_command_receipt_epic_seq_idx', - 'j5_a2a_comm_event_received_correlation_idx' + 'j5_a2a_comm_event_received_correlation_idx', + 'j5_a2a_comm_event_command_idx', + 'j5_a2a_exchange_open_pair_idx', + 'j5_a2a_exchange_id_idx', + 'j5_a2a_delivery_drain_idx', + 'j5_a2a_delivery_message_sender_idx', + 'j5_a2a_delivery_one_reply_idx' ) ORDER BY name `; @@ -80,14 +92,45 @@ it.effect("creates the exact namespaced ledger schema and receiver correlation c assert.deepStrictEqual(tables, [ { name: "j5_a2a_comm_command_receipt" }, { name: "j5_a2a_comm_event" }, + { name: "j5_a2a_delivery" }, { name: "j5_a2a_epic" }, { name: "j5_a2a_epic_membership" }, + { name: "j5_a2a_exchange" }, + { name: "j5_a2a_human_inbox_data" }, ]); - assert.equal(indexes[0]?.name, "j5_a2a_comm_command_receipt_epic_seq_idx"); - assert.include(indexes[0]?.sql ?? "", "ON j5_a2a_comm_command_receipt(epic_id, result_seq)"); - assert.equal(indexes[1]?.name, "j5_a2a_comm_event_received_correlation_idx"); - assert.include(indexes[1]?.sql ?? "", "ON j5_a2a_comm_event(epic_id, correlation_id)"); - assert.include(indexes[1]?.sql ?? "", "WHERE kind = 'message.received'"); + const indexesByName = new Map(indexes.map((index) => [index.name, index.sql])); + assert.include( + indexesByName.get("j5_a2a_comm_command_receipt_epic_seq_idx") ?? "", + "ON j5_a2a_comm_command_receipt(epic_id, result_seq)", + ); + assert.include( + indexesByName.get("j5_a2a_comm_event_received_correlation_idx") ?? "", + "WHERE kind = 'message.received'", + ); + assert.include( + indexesByName.get("j5_a2a_comm_event_command_idx") ?? "", + "ON j5_a2a_comm_event(command_id, epic_id, seq)", + ); + assert.include( + indexesByName.get("j5_a2a_exchange_open_pair_idx") ?? "", + "WHERE status = 'open'", + ); + assert.include( + indexesByName.get("j5_a2a_exchange_id_idx") ?? "", + "ON j5_a2a_exchange(exchange_id)", + ); + assert.include( + indexesByName.get("j5_a2a_delivery_drain_idx") ?? "", + "ON j5_a2a_delivery(status, next_attempt_at, sent_seq)", + ); + assert.include( + indexesByName.get("j5_a2a_delivery_message_sender_idx") ?? "", + "ON j5_a2a_delivery(message_id, sender_id)", + ); + assert.include( + indexesByName.get("j5_a2a_delivery_one_reply_idx") ?? "", + "WHERE exchange_id IS NOT NULL AND exchange_role = 'reply'", + ); assert.deepStrictEqual(unprefixed, []); }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), ); diff --git a/apps/server/src/j5/a2a/Migrations.ts b/apps/server/src/j5/a2a/Migrations.ts index 2e4db65ad737d..e8debdc8a26cd 100644 --- a/apps/server/src/j5/a2a/Migrations.ts +++ b/apps/server/src/j5/a2a/Migrations.ts @@ -2,10 +2,14 @@ import * as Effect from "effect/Effect"; import * as Migrator from "effect/unstable/sql/Migrator"; import Migration0001 from "./migrations/001_EpicCommunicationLedger.ts"; +import Migration0002 from "./migrations/002_SendDeliverReply.ts"; export const J5_A2A_MIGRATIONS_TABLE = "j5_a2a_migrations"; -export const migrationEntries = [[1, "EpicCommunicationLedger", Migration0001]] as const; +export const migrationEntries = [ + [1, "EpicCommunicationLedger", Migration0001], + [2, "SendDeliverReply", Migration0002], +] as const; const makeMigrationLoader = (throughId?: number) => Migrator.fromRecord( diff --git a/apps/server/src/j5/a2a/README.md b/apps/server/src/j5/a2a/README.md new file mode 100644 index 0000000000000..eb6dcf56950c5 --- /dev/null +++ b/apps/server/src/j5/a2a/README.md @@ -0,0 +1,13 @@ +# J5 A2A runtime configuration + +`envelopes.v1.json` is the human-owned wording source for every injected A2A envelope and the two MCP tool descriptions. Templates use literal `{{name}}` placeholders; `EnvelopeFormatter.ts` is the only renderer. Change wording in the JSON file, keep every referenced placeholder, and bump `version` when the rendered contract changes. + +`delivery-config.v1.json` owns the retry backoff and alarm threshold. Attempts reuse one upstream command/message id pair derived from the durable ledger message id. Increasing an attempt never rotates those ids: a permanently rejected upstream receipt must become a visible alarm rather than risk a second injection. + +Production membership provisioning is intentionally absent from A2. Native threads without an internally registered home epic are not A2A participants. The authenticated MCP toolkit exposes only `send_message` and `list_participants`; both fail closed when the caller has no provisioned epic membership. It does not expose an agent-invocable join, default epic creation, or epic movement. The membership schema, projection, and lifecycle events remain as the durable foundation for provisioning outside the agent tool surface. + +The named coordinated **home-epic registrar + A6 creation integrations** follow-up owns internal creation-time registration and its product integrations: users create and choose epics, and a spawned agent inherits its spawner's home epic. That follow-up also owns the executable live-proof runbook and a fresh real Codex-to-Claude proof before A3 is staffed. This PR does not infer those creation-time seams early or advertise a tool that production cannot provision truthfully. + +For a cross-epic send, the receiver ledger's idempotent `message.received` records durable acceptance of the sender's act before transport is attempted. It does not claim successful thread injection: delivery success, retries, and the terminal alarm remain in the sender epic's delivery projection. + +Byte-equivalent rebuilds for the A2 exchange and delivery projections are deferred to the measured-projections milestone (M5). A2 keeps those tables derivable from the communication ledger but does not expose their rebuild operation. diff --git a/apps/server/src/j5/a2a/SendService.test.ts b/apps/server/src/j5/a2a/SendService.test.ts new file mode 100644 index 0000000000000..160a9bb84dbe2 --- /dev/null +++ b/apps/server/src/j5/a2a/SendService.test.ts @@ -0,0 +1,409 @@ +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 { A2ALedger, layer as ledgerLayer } from "./LedgerService.ts"; +import { runJ5A2AMigrations } from "./Migrations.ts"; +import { A2ASendService, layer as sendLayer } from "./SendService.ts"; +import { CommCommandId, EpicId, ParticipantId, type AgentParticipant } from "./contracts.ts"; + +const timestamp = "2026-08-16T12:00:00.000Z"; + +const database = NodeSqliteClient.layerMemory(); +const ledger = ledgerLayer.pipe(Layer.provide(database)); +const send = sendLayer.pipe(Layer.provide(ledger), Layer.provide(database)); +const testLayer = Layer.mergeAll(database, ledger, send); + +const sender: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:sender"), + threadId: ThreadId.make("thread:sender"), +}; +const receiver: AgentParticipant = { + kind: "agent", + id: ParticipantId.make("agent:receiver"), + threadId: ThreadId.make("thread:receiver"), +}; + +const setupSameEpic = Effect.fn("test.j5.a2a.setupSameEpic")(function* () { + yield* runJ5A2AMigrations(); + const ledgerService = yield* A2ALedger; + const epicId = EpicId.make("epic:exchange"); + yield* ledgerService.createEpic({ + epic: { id: epicId, name: "Exchange", createdAt: timestamp }, + }); + for (const [index, participant] of [sender, receiver].entries()) { + yield* ledgerService.append({ + commandId: CommCommandId.make(`command:join:${index}`), + epicId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: participant.id, + exchangeId: null, + correlationId: null, + payload: { participant }, + createdAt: timestamp, + }, + }); + } + return epicId; +}); + +it.effect("opens once per sender-receiver pair, joins follow-ups, and one reply closes", () => + Effect.gen(function* () { + const epicId = yield* setupSameEpic(); + const service = yield* A2ASendService; + const sql = yield* SqlClient.SqlClient; + + const first = yield* service.send({ + commandId: CommCommandId.make("command:exchange:first"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Can you verify delivery?", + expectReply: true, + intent: "Verify the delivery path", + acceptedAt: timestamp, + }); + const followup = yield* service.send({ + commandId: CommCommandId.make("command:exchange:followup"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Please include the crash window.", + expectReply: true, + acceptedAt: timestamp, + }); + assert.equal(followup.exchangeId, first.exchangeId); + assert.isTrue(followup.joinedExistingExchange); + + const reply = yield* service.send({ + commandId: CommCommandId.make("command:exchange:reply"), + senderThreadId: receiver.threadId, + to: sender.id, + message: "Verified.", + exchangeId: first.exchangeId!, + acceptedAt: timestamp, + }); + assert.equal(reply.exchangeState, "closed"); + assert.deepStrictEqual( + yield* service.send({ + commandId: CommCommandId.make("command:exchange:reply"), + senderThreadId: receiver.threadId, + to: sender.id, + message: "Verified.", + exchangeId: first.exchangeId!, + acceptedAt: timestamp, + }), + reply, + "the same-epic reply command replays its original durable sequence", + ); + assert.deepStrictEqual( + yield* service.send({ + commandId: CommCommandId.make("command:exchange:first"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Can you verify delivery?", + expectReply: true, + intent: "Verify the delivery path", + acceptedAt: timestamp, + }), + first, + "the opening command replays its original result after closure", + ); + assert.deepStrictEqual( + yield* service.send({ + commandId: CommCommandId.make("command:exchange:followup"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Please include the crash window.", + expectReply: true, + acceptedAt: timestamp, + }), + followup, + "the follow-up command replays its original result after closure", + ); + + const rows = yield* sql<{ readonly kind: string; readonly count: number }>` + SELECT kind, COUNT(*) AS count + FROM j5_a2a_comm_event + WHERE epic_id = ${epicId} + AND kind IN ('exchange.opened', 'message.sent', 'exchange.closed') + GROUP BY kind + ORDER BY kind + `; + assert.deepStrictEqual(rows, [ + { kind: "exchange.closed", count: 1 }, + { kind: "exchange.opened", count: 1 }, + { kind: "message.sent", count: 3 }, + ]); + + const closedError = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:exchange:second-reply"), + senderThreadId: receiver.threadId, + to: sender.id, + message: "A duplicate reply.", + exchangeId: first.exchangeId!, + acceptedAt: timestamp, + }), + ); + assert.equal(closedError._tag, "A2AExchangeNotOpenError"); + }).pipe(Effect.provide(testLayer)), +); + +it.effect("validates intent and human-only urgency at exchange open", () => + Effect.gen(function* () { + const epicId = yield* setupSameEpic(); + const ledgerService = yield* A2ALedger; + const service = yield* A2ASendService; + yield* ledgerService.append({ + commandId: CommCommandId.make("command:join:human"), + epicId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: ParticipantId.make("human:global"), + exchangeId: null, + correlationId: null, + payload: { participant: { kind: "human" } }, + createdAt: timestamp, + }, + }); + + const missingIntent = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:missing-intent"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Question", + expectReply: true, + acceptedAt: timestamp, + }), + ); + assert.equal(missingIntent._tag, "A2AIntentRequiredError"); + + const missingUrgency = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:missing-urgency"), + senderThreadId: sender.threadId, + to: ParticipantId.make("human:global"), + message: "Human question", + expectReply: true, + intent: "Obtain a human ruling", + acceptedAt: timestamp, + }), + ); + assert.equal(missingUrgency._tag, "A2AUrgencyRequiredError"); + + const wrongUrgency = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:wrong-urgency"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "Agent question", + expectReply: true, + intent: "Ask an agent", + urgency: "soon", + acceptedAt: timestamp, + }), + ); + assert.equal(wrongUrgency._tag, "A2AUrgencyNotAcceptedError"); + + const oneShotUrgency = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:one-shot-urgency"), + senderThreadId: sender.threadId, + to: ParticipantId.make("human:global"), + message: "One-shot human message", + urgency: "fyi", + acceptedAt: timestamp, + }), + ); + assert.equal(oneShotUrgency._tag, "A2AUrgencyRequiresExchangeError"); + }).pipe(Effect.provide(testLayer)), +); + +it.effect("rolls back the send receipt when its projection write fails", () => + Effect.gen(function* () { + yield* setupSameEpic(); + const service = yield* A2ASendService; + const sql = yield* SqlClient.SqlClient; + const command = CommCommandId.make("command:receipt-rollback"); + yield* sql` + CREATE TRIGGER j5_a2a_test_fail_exchange_projection + BEFORE INSERT ON j5_a2a_exchange + WHEN NEW.exchange_id LIKE 'exchange:j5:a2a:%' + BEGIN + SELECT RAISE(ABORT, 'forced projection failure'); + END + `; + + yield* Effect.flip( + service.send({ + commandId: command, + senderThreadId: sender.threadId, + to: receiver.id, + message: "This transaction must roll back.", + expectReply: true, + intent: "Prove receipt rollback", + acceptedAt: timestamp, + }), + ); + const poisoned = yield* sql<{ readonly receipts: number; readonly events: number }>` + SELECT + (SELECT COUNT(*) FROM j5_a2a_comm_command_receipt WHERE command_id = ${command}) AS receipts, + (SELECT COUNT(*) FROM j5_a2a_comm_event WHERE command_id = ${command}) AS events + `; + assert.deepStrictEqual(poisoned, [{ receipts: 0, events: 0 }]); + + yield* sql`DROP TRIGGER j5_a2a_test_fail_exchange_projection`; + const retry = yield* service.send({ + commandId: command, + senderThreadId: sender.threadId, + to: receiver.id, + message: "This transaction must roll back.", + expectReply: true, + intent: "Prove receipt rollback", + acceptedAt: timestamp, + }); + assert.equal(retry.exchangeState, "open"); + }).pipe(Effect.provide(testLayer)), +); + +it.effect("fails closed when a native thread has no provisioned epic membership", () => + Effect.gen(function* () { + yield* runJ5A2AMigrations(); + const service = yield* A2ASendService; + const sql = yield* SqlClient.SqlClient; + const nativeThreadId = ThreadId.make("thread:native-without-home-epic"); + + const listError = yield* Effect.flip(service.listParticipants(nativeThreadId)); + assert.equal(listError._tag, "A2ASenderNotJoinedError"); + assert.include(listError.message, "native thread"); + assert.include(listError.message, "no registered home epic"); + assert.include(listError.message, "wrapper-spawned agent"); + assert.include(listError.message, "controlled test seeding"); + assert.include(listError.message, "home-epic registrar + A6 creation integrations follow-up"); + assert.include(listError.message, "Stop this messaging attempt"); + assert.notMatch(listError.message, /ask the user|product workflow|list_participants again/i); + + const sendError = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:native-without-home-epic"), + senderThreadId: nativeThreadId, + to: receiver.id, + message: "This must fail without provisioning.", + acceptedAt: timestamp, + }), + ); + assert.equal(sendError._tag, "A2ASenderNotJoinedError"); + + const state = yield* sql<{ readonly epics: number; readonly events: number }>` + SELECT + (SELECT COUNT(*) FROM j5_a2a_epic) AS epics, + (SELECT COUNT(*) FROM j5_a2a_comm_event) AS events + `; + assert.deepStrictEqual(state, [{ epics: 0, events: 0 }]); + }).pipe(Effect.provide(testLayer)), +); + +it.effect("lists membership-derived participant capabilities", () => + Effect.gen(function* () { + const epicId = yield* setupSameEpic(); + yield* (yield* A2ALedger).append({ + commandId: CommCommandId.make("command:list:join:human"), + epicId, + acceptedAt: timestamp, + event: { + kind: "participant.joined", + sender: null, + receiver: ParticipantId.make("human:global"), + exchangeId: null, + correlationId: null, + payload: { participant: { kind: "human" } }, + createdAt: timestamp, + }, + }); + const rows = yield* (yield* A2ASendService).listParticipants(sender.threadId); + assert.deepStrictEqual( + rows.map((row) => ({ + id: row.participantId, + canReceiveMessage: row.canReceiveMessage, + canOpenExchange: row.canOpenExchange, + acceptsUrgency: row.acceptsUrgency, + })), + [ + { + id: receiver.id, + canReceiveMessage: true, + canOpenExchange: true, + acceptsUrgency: false, + }, + { + id: sender.id, + canReceiveMessage: true, + canOpenExchange: true, + acceptsUrgency: false, + }, + { + id: ParticipantId.make("human:global"), + canReceiveMessage: true, + canOpenExchange: true, + acceptsUrgency: true, + }, + ], + ); + }).pipe(Effect.provide(testLayer)), +); + +it.effect("marks ambiguous participant rows unavailable before send", () => + Effect.gen(function* () { + yield* setupSameEpic(); + const ledgerService = yield* A2ALedger; + const duplicateEpicId = EpicId.make("epic:exchange:duplicate-receiver"); + yield* ledgerService.createEpic({ + epic: { id: duplicateEpicId, name: "Duplicate receiver", createdAt: timestamp }, + }); + yield* ledgerService.appendEvents({ + commandId: CommCommandId.make("command:join:duplicate-receiver"), + epicId: duplicateEpicId, + acceptedAt: timestamp, + events: [ + { + kind: "participant.joined", + sender: null, + receiver: receiver.id, + exchangeId: null, + correlationId: null, + payload: { participant: receiver }, + createdAt: timestamp, + }, + ], + }); + + const service = yield* A2ASendService; + const rows = (yield* service.listParticipants(sender.threadId)).filter( + (row) => row.participantId === receiver.id, + ); + assert.lengthOf(rows, 2); + assert.isTrue(rows.every((row) => !row.canReceiveMessage && !row.canOpenExchange)); + + const error = yield* Effect.flip( + service.send({ + commandId: CommCommandId.make("command:ambiguous-receiver"), + senderThreadId: sender.threadId, + to: receiver.id, + message: "This must fail before append.", + acceptedAt: timestamp, + }), + ); + assert.equal(error._tag, "A2AAmbiguousParticipantError"); + assert.include(error.message, "choose a participantId with canReceiveMessage=true"); + assert.include(error.message, "ask the human"); + }).pipe(Effect.provide(testLayer)), +); diff --git a/apps/server/src/j5/a2a/SendService.ts b/apps/server/src/j5/a2a/SendService.ts new file mode 100644 index 0000000000000..6d8af8ff7beb1 --- /dev/null +++ b/apps/server/src/j5/a2a/SendService.ts @@ -0,0 +1,474 @@ +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 type { SqlError } from "effect/unstable/sql/SqlError"; + +import { + CommCommandId, + type CommEvent, + CorrelationId, + EpicId, + ExchangeId, + GLOBAL_HUMAN_PARTICIPANT_ID, + LedgerMessageId, + Participant, + type ParticipantDirectoryRow, + type ParticipantId, + type SendMessageInput, + type SendMessageResult, + participantId, +} from "./contracts.ts"; +import { A2ALedger, type A2ALedgerError } from "./LedgerService.ts"; + +export class A2ASenderNotJoinedError extends Schema.TaggedErrorClass()( + "A2ASenderNotJoinedError", + { threadId: Schema.String }, +) { + override get message(): string { + return `Cross-agent messaging is unavailable for native thread ${this.threadId} because it has no registered home epic. Participation currently requires a wrapper-spawned agent that already has a home epic or controlled test seeding. Native user-created home provisioning is deferred to the home-epic registrar + A6 creation integrations follow-up. Stop this messaging attempt.`; + } +} + +export class A2AParticipantNotFoundError extends Schema.TaggedErrorClass()( + "A2AParticipantNotFoundError", + { participantId: Schema.String }, +) { + override get message(): string { + return `Participant ${this.participantId} is not currently reachable. Call list_participants and choose a row with canReceiveMessage=true.`; + } +} + +export class A2AAmbiguousParticipantError extends Schema.TaggedErrorClass()( + "A2AAmbiguousParticipantError", + { participantId: Schema.String }, +) { + override get message(): string { + return `Participant ${this.participantId} is active in more than one epic and cannot be addressed unambiguously. Call list_participants and choose a participantId with canReceiveMessage=true, or ask the human to repair epic membership.`; + } +} + +export class A2AIntentRequiredError extends Schema.TaggedErrorClass()( + "A2AIntentRequiredError", + {}, +) { + override get message(): string { + return "Opening an exchange requires intent. Retry send_message with a one-line intent summary."; + } +} + +export class A2AUrgencyRequiredError extends Schema.TaggedErrorClass()( + "A2AUrgencyRequiredError", + {}, +) { + override get message(): string { + return "Opening an exchange to the human requires urgency=blocking|soon|fyi. Retry send_message with urgency."; + } +} + +export class A2AUrgencyNotAcceptedError extends Schema.TaggedErrorClass()( + "A2AUrgencyNotAcceptedError", + { participantId: Schema.String }, +) { + override get message(): string { + return `Participant ${this.participantId} does not accept urgency. Retry send_message without urgency.`; + } +} + +export class A2AUrgencyRequiresExchangeError extends Schema.TaggedErrorClass()( + "A2AUrgencyRequiresExchangeError", + {}, +) { + override get message(): string { + return "Urgency applies only when opening a reply-expected exchange to the human. Retry without urgency, or set expect_reply=true with intent and urgency."; + } +} + +export class A2AExchangeNotOpenError extends Schema.TaggedErrorClass()( + "A2AExchangeNotOpenError", + { exchangeId: Schema.String }, +) { + override get message(): string { + return `Exchange ${this.exchangeId} is not open. Call send_message without exchange_id to start a new message or exchange.`; + } +} + +export class A2AExchangeParticipantMismatchError extends Schema.TaggedErrorClass()( + "A2AExchangeParticipantMismatchError", + { + exchangeId: Schema.String, + senderId: Schema.String, + receiverId: Schema.String, + }, +) { + override get message(): string { + return `Exchange ${this.exchangeId} does not connect ${this.senderId} to ${this.receiverId}. Call list_participants and use the exchange's original peer.`; + } +} + +export class A2AExchangeAlreadyAnsweredError extends Schema.TaggedErrorClass()( + "A2AExchangeAlreadyAnsweredError", + { exchangeId: Schema.String }, +) { + override get message(): string { + return `Exchange ${this.exchangeId} already has its one durable reply and is closing or closed. Call send_message without exchange_id to start a new message or exchange.`; + } +} + +export type A2ASendError = + | A2ALedgerError + | Schema.SchemaError + | SqlError + | A2ASenderNotJoinedError + | A2AParticipantNotFoundError + | A2AAmbiguousParticipantError + | A2AIntentRequiredError + | A2AUrgencyRequiredError + | A2AUrgencyNotAcceptedError + | A2AUrgencyRequiresExchangeError + | A2AExchangeNotOpenError + | A2AExchangeAlreadyAnsweredError + | A2AExchangeParticipantMismatchError; + +interface MembershipRow { + readonly epic_id: string; + readonly participant_id: string; + readonly thread_id: string | null; + readonly payload: string; +} + +interface ExchangeRow { + readonly epic_id: string; + readonly exchange_id: string; + readonly sender_id: string; + readonly receiver_id: string; + readonly status: "open" | "closed"; +} + +interface ExistingMessageRow { + readonly epic_id: string; + readonly sender_id: string; + readonly receiver_id: string; + readonly exchange_id: string | null; + readonly exchange_role: "none" | "ask" | "followup" | "reply"; + readonly sent_seq: number; +} + +const decodeParticipant = Schema.decodeUnknownEffect(Schema.fromJsonString(Participant)); + +const messageIdFor = (commandId: CommCommandId) => + LedgerMessageId.make(`message:j5:a2a:${encodeURIComponent(commandId)}`); + +const exchangeIdFor = (commandId: CommCommandId) => + ExchangeId.make(`exchange:j5:a2a:${encodeURIComponent(commandId)}`); + +const correlationIdFor = (commandId: CommCommandId) => + CorrelationId.make(`correlation:j5:a2a:${encodeURIComponent(commandId)}`); + +export interface A2ASendServiceShape { + readonly send: (input: SendMessageInput) => Effect.Effect; + readonly listParticipants: ( + senderThreadId: ThreadId, + ) => Effect.Effect, A2ASendError>; +} + +export class A2ASendService extends Context.Service()( + "t3/j5/a2a/SendService/A2ASendService", +) {} + +export const layer: Layer.Layer = + Layer.effect( + A2ASendService, + Effect.gen(function* () { + const ledger = yield* A2ALedger; + const sql = yield* SqlClient.SqlClient; + + const membershipRows = Effect.fn("j5.a2a.send.membershipRows")(function* () { + return yield* sql` + SELECT epic_id, participant_id, thread_id, payload + FROM j5_a2a_epic_membership + ORDER BY epic_id, participant_id + `; + }); + + const senderMembership = Effect.fn("j5.a2a.send.senderMembership")(function* ( + threadId: ThreadId, + ) { + const matches = (yield* membershipRows()).filter((row) => row.thread_id === threadId); + if (matches.length !== 1) { + return yield* new A2ASenderNotJoinedError({ threadId }); + } + const row = matches[0]!; + return { + epicId: EpicId.make(row.epic_id), + participantId: row.participant_id as ParticipantId, + }; + }); + + const participantMembership = Effect.fn("j5.a2a.send.participantMembership")(function* ( + id: ParticipantId, + senderEpicId: EpicId, + ) { + const matches = (yield* membershipRows()).filter((row) => row.participant_id === id); + if (id === GLOBAL_HUMAN_PARTICIPANT_ID) { + const local = matches.find((row) => row.epic_id === senderEpicId); + if (local === undefined) { + return yield* new A2AParticipantNotFoundError({ participantId: id }); + } + return { epicId: senderEpicId, participant: yield* decodeParticipant(local.payload) }; + } + if (matches.length === 0) { + return yield* new A2AParticipantNotFoundError({ participantId: id }); + } + if (matches.length > 1) { + return yield* new A2AAmbiguousParticipantError({ participantId: id }); + } + return { + epicId: EpicId.make(matches[0]!.epic_id), + participant: yield* decodeParticipant(matches[0]!.payload), + }; + }); + + const listParticipants: A2ASendServiceShape["listParticipants"] = (senderThreadId) => + Effect.gen(function* () { + const sender = yield* senderMembership(senderThreadId); + const rows = yield* membershipRows(); + const membershipCounts = new Map(); + for (const row of rows) { + membershipCounts.set( + row.participant_id, + (membershipCounts.get(row.participant_id) ?? 0) + 1, + ); + } + const selected = rows.filter( + (row) => + row.participant_id !== GLOBAL_HUMAN_PARTICIPANT_ID || row.epic_id === sender.epicId, + ); + return yield* Effect.forEach( + selected, + (row) => + decodeParticipant(row.payload).pipe( + Effect.map((participant) => { + const id = participantId(participant); + const addressable = + id === GLOBAL_HUMAN_PARTICIPANT_ID || membershipCounts.get(id) === 1; + return { + epicId: EpicId.make(row.epic_id), + participantId: id, + participant, + canReceiveMessage: addressable, + canOpenExchange: addressable, + acceptsUrgency: participant.kind === "human", + }; + }), + ), + { concurrency: 1 }, + ); + }); + + const replayedSend = Effect.fn("j5.a2a.send.replayedSend")(function* ( + messageId: LedgerMessageId, + senderId: ParticipantId, + ) { + const rows = yield* sql` + SELECT + epic_id, + sender_id, + receiver_id, + exchange_id, + exchange_role, + sent_seq + FROM j5_a2a_delivery + WHERE message_id = ${messageId} + AND sender_id = ${senderId} + LIMIT 2 + `; + if (rows.length !== 1) return null; + const row = rows[0]!; + const exchange = + row.exchange_id === null + ? [] + : yield* sql` + SELECT epic_id, exchange_id, sender_id, receiver_id, status + FROM j5_a2a_exchange + WHERE exchange_id = ${row.exchange_id} + LIMIT 1 + `; + const isCrossEpicReply = + exchange[0] !== undefined && + exchange[0].epic_id !== row.epic_id && + exchange[0].receiver_id === row.sender_id && + exchange[0].sender_id === row.receiver_id; + return { + messageId, + exchangeId: row.exchange_id === null ? null : ExchangeId.make(row.exchange_id), + exchangeState: + row.exchange_role === "none" + ? ("none" as const) + : row.exchange_role !== "reply" + ? ("open" as const) + : isCrossEpicReply + ? ("closing" as const) + : ("closed" as const), + joinedExistingExchange: row.exchange_role === "followup", + durableAtSeq: row.sent_seq, + } satisfies SendMessageResult; + }); + + const send: A2ASendServiceShape["send"] = (input) => + Effect.gen(function* () { + const messageId = messageIdFor(input.commandId); + const sender = yield* senderMembership(input.senderThreadId); + const replay = yield* replayedSend(messageId, sender.participantId); + if (replay !== null) return replay; + + const receiver = yield* participantMembership(input.to, sender.epicId); + const receiverId = participantId(receiver.participant); + let exchangeId: ExchangeId | null = null; + let exchangeState: SendMessageResult["exchangeState"] = "none"; + let exchangeRole: "none" | "ask" | "followup" | "reply" = "none"; + let joinedExistingExchange = false; + let openEvent: CommEvent | undefined; + let closeEvent: CommEvent | undefined; + + if (input.exchangeId !== undefined) { + if (input.urgency !== undefined) { + return yield* new A2AUrgencyRequiresExchangeError(); + } + const rows = yield* sql` + SELECT epic_id, exchange_id, sender_id, receiver_id, status + FROM j5_a2a_exchange + WHERE exchange_id = ${input.exchangeId} + LIMIT 2 + `; + const exchange = rows.length === 1 ? rows[0] : undefined; + if (exchange === undefined || exchange.status !== "open") { + return yield* new A2AExchangeNotOpenError({ exchangeId: input.exchangeId }); + } + const isFollowup = + exchange.sender_id === sender.participantId && exchange.receiver_id === receiverId; + const isReply = + exchange.receiver_id === sender.participantId && exchange.sender_id === receiverId; + if (!isFollowup && !isReply) { + return yield* new A2AExchangeParticipantMismatchError({ + exchangeId: input.exchangeId, + senderId: sender.participantId, + receiverId, + }); + } + exchangeId = input.exchangeId; + joinedExistingExchange = isFollowup; + if (isReply) { + const acceptedReplies = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM j5_a2a_delivery + WHERE exchange_id = ${exchangeId} AND exchange_role = 'reply' + `; + if ((acceptedReplies[0]?.count ?? 0) > 0) { + return yield* new A2AExchangeAlreadyAnsweredError({ exchangeId }); + } + exchangeRole = "reply"; + exchangeState = exchange.epic_id === sender.epicId ? "closed" : "closing"; + if (exchange.epic_id === sender.epicId) { + closeEvent = { + kind: "exchange.closed", + sender: sender.participantId, + receiver: receiverId, + exchangeId, + correlationId: correlationIdFor(input.commandId), + payload: { replyMessageId: messageId }, + createdAt: input.acceptedAt, + }; + } + } else { + exchangeRole = "followup"; + exchangeState = "open"; + } + } else if (input.expectReply === true) { + const existing = yield* sql` + SELECT epic_id, exchange_id, sender_id, receiver_id, status + FROM j5_a2a_exchange + WHERE epic_id = ${sender.epicId} + AND sender_id = ${sender.participantId} + AND receiver_id = ${receiverId} + AND status = 'open' + LIMIT 1 + `; + if (existing[0] !== undefined) { + exchangeId = ExchangeId.make(existing[0].exchange_id); + joinedExistingExchange = true; + exchangeRole = "followup"; + } else { + if (input.intent === undefined) return yield* new A2AIntentRequiredError(); + if (receiver.participant.kind === "human" && input.urgency === undefined) { + return yield* new A2AUrgencyRequiredError(); + } + if (receiver.participant.kind !== "human" && input.urgency !== undefined) { + return yield* new A2AUrgencyNotAcceptedError({ participantId: receiverId }); + } + exchangeId = exchangeIdFor(input.commandId); + exchangeRole = "ask"; + openEvent = { + kind: "exchange.opened", + sender: sender.participantId, + receiver: receiverId, + exchangeId, + correlationId: correlationIdFor(input.commandId), + payload: { + intent: input.intent, + urgency: input.urgency ?? null, + }, + createdAt: input.acceptedAt, + }; + } + exchangeState = "open"; + } else if (input.urgency !== undefined) { + return yield* new A2AUrgencyRequiresExchangeError(); + } + + const correlationId = correlationIdFor(input.commandId); + const result = yield* ledger.appendEvents({ + commandId: input.commandId, + epicId: sender.epicId, + acceptedAt: input.acceptedAt, + events: [ + ...(openEvent === undefined ? [] : [openEvent]), + { + kind: "message.sent", + sender: sender.participantId, + receiver: receiverId, + exchangeId, + correlationId, + payload: { + messageId, + text: input.message, + originEpicId: sender.epicId, + receiverEpicId: receiver.epicId, + exchangeRole, + }, + createdAt: input.acceptedAt, + }, + ...(closeEvent === undefined ? [] : [closeEvent]), + ], + }); + const sent = result.events.find((event) => event.kind === "message.sent"); + if (sent === undefined) { + return yield* new A2AParticipantNotFoundError({ participantId: receiverId }); + } + const opened = result.events.some((event) => event.kind === "exchange.opened"); + const closed = result.events.some((event) => event.kind === "exchange.closed"); + return { + messageId, + exchangeId, + exchangeState: closed ? "closed" : exchangeState, + joinedExistingExchange: + exchangeId !== null && !opened && !closed ? joinedExistingExchange : false, + durableAtSeq: sent.seq, + } satisfies SendMessageResult; + }); + + return A2ASendService.of({ send, listParticipants }); + }), + ); diff --git a/apps/server/src/j5/a2a/contracts.ts b/apps/server/src/j5/a2a/contracts.ts index 99f5b021123cb..91191257838d4 100644 --- a/apps/server/src/j5/a2a/contracts.ts +++ b/apps/server/src/j5/a2a/contracts.ts @@ -23,6 +23,12 @@ export type ParticipantId = typeof ParticipantId.Type; export const CommCommandId = Identifier.pipe(Schema.brand("J5A2ACommCommandId")); export type CommCommandId = typeof CommCommandId.Type; +export const LedgerMessageId = Identifier.pipe(Schema.brand("J5A2ALedgerMessageId")); +export type LedgerMessageId = typeof LedgerMessageId.Type; + +export const Urgency = Schema.Literals(["blocking", "soon", "fyi"]); +export type Urgency = typeof Urgency.Type; + export const GLOBAL_HUMAN_PARTICIPANT_ID = ParticipantId.make("human:global"); export const AgentParticipant = Schema.Struct({ @@ -156,6 +162,98 @@ export const CommCommandReceipt = Schema.Struct({ }); export type CommCommandReceipt = typeof CommCommandReceipt.Type; +export const AppendCommEventsCommand = Schema.Struct({ + commandId: CommCommandId, + epicId: EpicId, + acceptedAt: Schema.String, + events: Schema.Array(CommEvent).pipe(Schema.check(Schema.isMinLength(1))), +}); +export type AppendCommEventsCommand = typeof AppendCommEventsCommand.Type; + +export const ExchangeOpenedPayload = Schema.Struct({ + intent: Schema.String.check(Schema.isNonEmpty()), + urgency: Schema.NullOr(Urgency), +}); +export type ExchangeOpenedPayload = typeof ExchangeOpenedPayload.Type; + +export const MessageSentPayload = Schema.Struct({ + messageId: LedgerMessageId, + text: Schema.String.check(Schema.isNonEmpty()), + originEpicId: EpicId, + receiverEpicId: EpicId, + exchangeRole: Schema.Literals(["none", "ask", "followup", "reply"]), +}); +export type MessageSentPayload = typeof MessageSentPayload.Type; + +export const MessageDeliveredPayload = Schema.Struct({ + messageId: LedgerMessageId, + attempt: PositiveInt, + channel: Schema.Literals(["agent", "human"]), +}); +export type MessageDeliveredPayload = typeof MessageDeliveredPayload.Type; + +export const MessageDeliveryFailedPayload = Schema.Struct({ + messageId: LedgerMessageId, + attempt: PositiveInt, + error: Schema.String.check(Schema.isNonEmpty()), + nextAttemptAt: Schema.NullOr(Schema.String), + alarmed: Schema.Boolean, +}); +export type MessageDeliveryFailedPayload = typeof MessageDeliveryFailedPayload.Type; + +export const ExchangeClosedPayload = Schema.Struct({ + replyMessageId: LedgerMessageId, +}); +export type ExchangeClosedPayload = typeof ExchangeClosedPayload.Type; + +export const SendMessageInput = Schema.Struct({ + commandId: CommCommandId, + senderThreadId: ThreadId, + to: ParticipantId, + message: Schema.String.check(Schema.isNonEmpty()), + expectReply: Schema.optional(Schema.Boolean), + exchangeId: Schema.optional(ExchangeId), + intent: Schema.optional(Schema.String.check(Schema.isNonEmpty())), + urgency: Schema.optional(Urgency), + acceptedAt: Schema.String, +}); +export type SendMessageInput = typeof SendMessageInput.Type; + +export const SendMessageResult = Schema.Struct({ + messageId: LedgerMessageId, + exchangeId: Schema.NullOr(ExchangeId), + exchangeState: Schema.Literals(["none", "open", "closing", "closed"]), + joinedExistingExchange: Schema.Boolean, + durableAtSeq: PositiveInt, +}); +export type SendMessageResult = typeof SendMessageResult.Type; + +export const ParticipantDirectoryRow = Schema.Struct({ + epicId: EpicId, + participantId: ParticipantId, + participant: Participant, + canReceiveMessage: Schema.Boolean, + canOpenExchange: Schema.Boolean, + acceptsUrgency: Schema.Boolean, +}); +export type ParticipantDirectoryRow = typeof ParticipantDirectoryRow.Type; + +export const DeliveryAlarm = Schema.Struct({ + epicId: EpicId, + messageId: LedgerMessageId, + attempts: PositiveInt, + lastError: Schema.String, +}); +export type DeliveryAlarm = typeof DeliveryAlarm.Type; + +export const DeliveryMilestone = Schema.Struct({ + epicId: EpicId, + messageId: LedgerMessageId, + state: Schema.Literals(["delivered", "retry_scheduled", "alarmed"]), + attempt: PositiveInt, +}); +export type DeliveryMilestone = typeof DeliveryMilestone.Type; + export const LedgerCursor = Schema.Struct({ afterSeq: NonNegativeInt, snapshotEnd: Schema.optional(NonNegativeInt), diff --git a/apps/server/src/j5/a2a/delivery-config.v1.json b/apps/server/src/j5/a2a/delivery-config.v1.json new file mode 100644 index 0000000000000..71c4550055ce3 --- /dev/null +++ b/apps/server/src/j5/a2a/delivery-config.v1.json @@ -0,0 +1,6 @@ +{ + "version": 1, + "alarmAfterAttempts": 3, + "initialBackoffMs": 250, + "maximumBackoffMs": 5000 +} diff --git a/apps/server/src/j5/a2a/envelopes.v1.json b/apps/server/src/j5/a2a/envelopes.v1.json new file mode 100644 index 0000000000000..706c64b4b9bfa --- /dev/null +++ b/apps/server/src/j5/a2a/envelopes.v1.json @@ -0,0 +1,10 @@ +{ + "version": 5, + "peerMessage": "[Cross-agent message from {{senderId}} in epic {{originEpicId}}]\n\n{{message}}\n\n{{exchangeInstruction}}", + "humanMessage": "[Message from the human]\n\n{{message}}\n\nThe human is not watching this chat. They see only what you send back on this exchange.\n\n{{exchangeInstruction}}", + "silenceNotice": "[Cross-agent messaging system notice: {{noticeType}}]\n\n{{message}}\n\nThis is a platform-authored delivery signal, not a peer reply.", + "replyInstruction": "Reply once with send_message(to=\"{{senderId}}\", exchange_id=\"{{exchangeId}}\", message=\"...\") to close the exchange. Follow-ups from the asker carrying this id join the same exchange.", + "oneShotInstruction": "No reply is required. Use send_message without exchange_id only if a new message is needed.", + "sendToolDescription": "Durably send one message. client_request_id makes retries of the same logical call idempotent. expect_reply=true opens or joins one sender-to-receiver exchange and requires intent; a reply is send_message carrying exchange_id, and one reply closes that exchange. urgency is required only when opening an exchange to the human. The call returns after the sender ledger commit; delivery continues asynchronously. This tool is unavailable to a native thread without a registered home epic. Participation currently requires a wrapper-spawned agent that already has a home epic or controlled test seeding; native user-created home provisioning is deferred to the home-epic registrar + A6 creation integrations follow-up.", + "listToolDescription": "List reachable message recipients and the capabilities accepted by each row. Use this before send_message instead of discovering participant state through failures. This tool is unavailable to a native thread without a registered home epic. Participation currently requires a wrapper-spawned agent that already has a home epic or controlled test seeding; native user-created home provisioning is deferred to the home-epic registrar + A6 creation integrations follow-up." +} diff --git a/apps/server/src/j5/a2a/mcp/handlers.test.ts b/apps/server/src/j5/a2a/mcp/handlers.test.ts new file mode 100644 index 0000000000000..342308d628bca --- /dev/null +++ b/apps/server/src/j5/a2a/mcp/handlers.test.ts @@ -0,0 +1,81 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Sink from "effect/Sink"; +import * as Stream from "effect/Stream"; + +import { McpInvocationContext } from "../../../mcp/McpInvocationContext.ts"; +import { A2ADeliveryWorker } from "../DeliveryWorker.ts"; +import { A2ASendService } from "../SendService.ts"; +import { LedgerMessageId, ParticipantId, type SendMessageInput } from "../contracts.ts"; +import { J5ToolkitHandlersLive } from "./handlers.ts"; +import { J5Toolkit, type J5SendMessageInput } from "./tools.ts"; + +const invocation = { + environmentId: EnvironmentId.make("environment:j5:mcp-handler"), + threadId: ThreadId.make("thread:j5:mcp-handler"), + providerSessionId: "provider-session:j5:mcp-handler", + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["orchestration"] as const), + issuedAt: 1, +}; + +it.effect("derives send idempotency and sender identity from authenticated scope", () => + Effect.gen(function* () { + assert.deepStrictEqual(Object.keys(J5Toolkit.tools).sort(), [ + "list_participants", + "send_message", + ]); + const sends = yield* Ref.make>([]); + const participantId = ParticipantId.make("agent:j5:mcp-handler"); + const sendService = Layer.succeed( + A2ASendService, + A2ASendService.of({ + send: (input) => + Ref.update(sends, (items) => [...items, input]).pipe( + Effect.as({ + messageId: LedgerMessageId.make("message:j5:mcp-handler"), + exchangeId: null, + exchangeState: "none" as const, + joinedExistingExchange: false, + durableAtSeq: 1, + }), + ), + listParticipants: () => Effect.succeed([]), + }), + ); + const dependencies = Layer.mergeAll( + sendService, + Layer.mock(A2ADeliveryWorker)({ notify: Effect.void }), + NodeServices.layer, + ); + const layer = J5ToolkitHandlersLive.pipe(Layer.provideMerge(dependencies)); + + yield* Effect.gen(function* () { + const toolkit = yield* J5Toolkit; + const call = (args: J5SendMessageInput) => + toolkit + .handle("send_message", args) + .pipe( + Stream.unwrap, + Stream.run(Sink.last()), + Effect.flatMap(Effect.fromOption), + Effect.provideService(McpInvocationContext, invocation), + ); + const sendArguments = { + to: participantId, + message: "Idempotent MCP send", + client_request_id: "logical-send-1", + }; + yield* call(sendArguments); + yield* call(sendArguments); + const captured = yield* Ref.get(sends); + assert.lengthOf(captured, 2); + assert.equal(captured[0]?.commandId, captured[1]?.commandId); + assert.equal(captured[0]?.senderThreadId, invocation.threadId); + }).pipe(Effect.provide(layer)); + }), +); diff --git a/apps/server/src/j5/a2a/mcp/handlers.ts b/apps/server/src/j5/a2a/mcp/handlers.ts new file mode 100644 index 0000000000000..cfbc2dbedf72f --- /dev/null +++ b/apps/server/src/j5/a2a/mcp/handlers.ts @@ -0,0 +1,63 @@ +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { McpInvocationContext } from "../../../mcp/McpInvocationContext.ts"; +import { A2ADeliveryWorker } from "../DeliveryWorker.ts"; +import { A2ASendService } from "../SendService.ts"; +import { CommCommandId } from "../contracts.ts"; +import { J5Toolkit, type J5McpFailure } from "./tools.ts"; + +const failure = (error: unknown): J5McpFailure => ({ + code: + typeof error === "object" && error !== null && "_tag" in error + ? String(error._tag) + : "J5A2AError", + message: error instanceof Error ? error.message : String(error), +}); + +const stablePart = (value: string) => encodeURIComponent(value); + +export const commandIdForRequest = (input: { + readonly providerSessionId: string; + readonly requestKey: string; +}) => + CommCommandId.make( + `command:j5:a2a:mcp:${stablePart(input.providerSessionId)}:${stablePart(input.requestKey)}`, + ); + +const handlers = { + send_message: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const crypto = yield* Crypto.Crypto; + const service = yield* A2ASendService; + const worker = yield* A2ADeliveryWorker; + const acceptedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + const requestKey = input.client_request_id ?? (yield* crypto.randomUUIDv4); + const result = yield* service.send({ + commandId: commandIdForRequest({ + providerSessionId: scope.providerSessionId, + requestKey, + }), + senderThreadId: scope.threadId, + to: input.to, + message: input.message, + ...(input.expect_reply === undefined ? {} : { expectReply: input.expect_reply }), + ...(input.exchange_id === undefined ? {} : { exchangeId: input.exchange_id }), + ...(input.intent === undefined ? {} : { intent: input.intent }), + ...(input.urgency === undefined ? {} : { urgency: input.urgency }), + acceptedAt, + }); + yield* worker.notify; + return result; + }).pipe(Effect.mapError(failure)), + list_participants: () => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* A2ASendService; + return { participants: yield* service.listParticipants(scope.threadId) }; + }).pipe(Effect.mapError(failure)), +} satisfies Parameters[0]; + +export const J5ToolkitHandlersLive = J5Toolkit.toLayer(handlers); diff --git a/apps/server/src/j5/a2a/mcp/registration.ts b/apps/server/src/j5/a2a/mcp/registration.ts new file mode 100644 index 0000000000000..d37e6f4dd14ba --- /dev/null +++ b/apps/server/src/j5/a2a/mcp/registration.ts @@ -0,0 +1,16 @@ +import * as Layer from "effect/Layer"; +import { McpServer } from "effect/unstable/ai"; + +import { J5ToolkitHandlersLive } from "./handlers.ts"; +import { J5Toolkit } from "./tools.ts"; +import { J5A2ARuntimeLayer } from "../runtimeLayer.ts"; + +/** The single shared J5 MCP registration; later J5 milestones extend J5Toolkit only. */ +export const J5ToolkitRegistrationLive = McpServer.toolkit(J5Toolkit).pipe( + Layer.provide(J5ToolkitHandlersLive), +); + +/** Authenticated toolkit plus its independently drainable A2A runtime. */ +export const J5McpIntegrationLive = J5ToolkitRegistrationLive.pipe( + Layer.provideMerge(J5A2ARuntimeLayer), +); diff --git a/apps/server/src/j5/a2a/mcp/tools.ts b/apps/server/src/j5/a2a/mcp/tools.ts new file mode 100644 index 0000000000000..2ea7d7877d802 --- /dev/null +++ b/apps/server/src/j5/a2a/mcp/tools.ts @@ -0,0 +1,73 @@ +import { Tool, Toolkit } from "effect/unstable/ai"; +import * as Schema from "effect/Schema"; + +import * as Crypto from "effect/Crypto"; +import * as McpInvocationContext from "../../../mcp/McpInvocationContext.ts"; +import { A2A_LIST_TOOL_DESCRIPTION, A2A_SEND_TOOL_DESCRIPTION } from "../EnvelopeFormatter.ts"; +import { A2ADeliveryWorker } from "../DeliveryWorker.ts"; +import { A2ASendService } from "../SendService.ts"; +import { + ExchangeId, + ParticipantDirectoryRow, + ParticipantId, + SendMessageResult, + Urgency, +} from "../contracts.ts"; + +export const J5McpFailure = Schema.Struct({ + code: Schema.String, + message: Schema.String, +}); +export type J5McpFailure = typeof J5McpFailure.Type; + +export const J5SendMessageInput = Schema.Struct({ + to: ParticipantId, + message: Schema.String.check(Schema.isNonEmpty()), + client_request_id: Schema.optional(Schema.String.check(Schema.isNonEmpty())), + expect_reply: Schema.optional(Schema.Boolean), + exchange_id: Schema.optional(ExchangeId), + intent: Schema.optional(Schema.String.check(Schema.isNonEmpty())), + urgency: Schema.optional(Urgency), +}); +export type J5SendMessageInput = typeof J5SendMessageInput.Type; + +export const J5ListParticipantsResult = Schema.Struct({ + participants: Schema.Array(ParticipantDirectoryRow), +}); + +const dependencies = [ + McpInvocationContext.McpInvocationContext, + A2ASendService, + A2ADeliveryWorker, + Crypto.Crypto, +]; + +export const J5SendMessageTool = Tool.make("send_message", { + description: A2A_SEND_TOOL_DESCRIPTION, + parameters: J5SendMessageInput, + success: SendMessageResult, + failure: J5McpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Send a cross-agent message") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, false) + .annotate(Tool.OpenWorld, true); + +export const J5ListParticipantsTool = Tool.make("list_participants", { + description: A2A_LIST_TOOL_DESCRIPTION, + success: J5ListParticipantsResult, + failure: J5McpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "List cross-agent messaging participants") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true) + .annotate(Tool.OpenWorld, false); + +/** Shared J5 toolkit bootstrap. Later J5 milestones append their tools here. */ +export const J5Toolkit = Toolkit.make(J5SendMessageTool, J5ListParticipantsTool); diff --git a/apps/server/src/j5/a2a/migrations/002_SendDeliverReply.ts b/apps/server/src/j5/a2a/migrations/002_SendDeliverReply.ts new file mode 100644 index 0000000000000..15c4d52890ed2 --- /dev/null +++ b/apps/server/src/j5/a2a/migrations/002_SendDeliverReply.ts @@ -0,0 +1,91 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql`ALTER TABLE j5_a2a_comm_event ADD COLUMN command_id TEXT`; + yield* sql` + CREATE INDEX j5_a2a_comm_event_command_idx + ON j5_a2a_comm_event(command_id, epic_id, seq) + WHERE command_id IS NOT NULL + `; + + yield* sql` + CREATE TABLE j5_a2a_exchange ( + epic_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')), + 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 (epic_id, exchange_id), + FOREIGN KEY (epic_id) REFERENCES j5_a2a_epic(id) ON DELETE CASCADE + ) + `; + yield* sql` + CREATE UNIQUE INDEX j5_a2a_exchange_open_pair_idx + ON j5_a2a_exchange(epic_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 ( + epic_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_epic_id TEXT NOT NULL, + exchange_id TEXT, + exchange_role TEXT NOT NULL CHECK (exchange_role IN ('none', 'ask', 'followup', 'reply')), + 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, + PRIMARY KEY (epic_id, message_id), + FOREIGN KEY (epic_id) REFERENCES j5_a2a_epic(id) ON DELETE CASCADE + ) + `; + 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_human_inbox_data ( + origin_epic_id TEXT NOT NULL, + message_id TEXT NOT NULL, + exchange_id TEXT, + sender_id TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (origin_epic_id, message_id), + FOREIGN KEY (origin_epic_id) REFERENCES j5_a2a_epic(id) ON DELETE CASCADE + ) + `; +}); diff --git a/apps/server/src/j5/a2a/runtimeLayer.test.ts b/apps/server/src/j5/a2a/runtimeLayer.test.ts new file mode 100644 index 0000000000000..ee94b02855704 --- /dev/null +++ b/apps/server/src/j5/a2a/runtimeLayer.test.ts @@ -0,0 +1,44 @@ +import { assert, it } from "@effect/vitest"; +import * as Context from "effect/Context"; +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 { ThreadManagementService } from "../../orchestration-v2/ThreadManagementService.ts"; +import { layer as ledgerLayer } from "./LedgerService.ts"; +import { runJ5A2AMigrations } from "./Migrations.ts"; +import { makeJ5A2ARuntimeLayer } from "./runtimeLayer.ts"; + +it.effect("shares one ledger and thread-management instance across the runtime graph", () => + Effect.scoped( + Effect.gen(function* () { + const databaseContext = yield* Layer.build(NodeSqliteClient.layerMemory()); + const database = Layer.succeed( + SqlClient.SqlClient, + Context.get(databaseContext, SqlClient.SqlClient), + ); + yield* runJ5A2AMigrations().pipe(Effect.provide(database)); + + let ledgerBuilds = 0; + let threadManagementBuilds = 0; + const countedLedger = ledgerLayer.pipe( + Layer.tap(() => Effect.sync(() => (ledgerBuilds += 1))), + ); + const countedThreadManagement = Layer.mock(ThreadManagementService)({}).pipe( + Layer.tap(() => Effect.sync(() => (threadManagementBuilds += 1))), + ); + const secondThreadConsumer = Layer.effectDiscard(ThreadManagementService.pipe(Effect.asVoid)); + const runtime = makeJ5A2ARuntimeLayer({ ledger: countedLedger }); + yield* Layer.build( + Layer.mergeAll(runtime, secondThreadConsumer).pipe( + Layer.provide(countedThreadManagement), + Layer.provide(database), + ), + ); + + assert.equal(ledgerBuilds, 1); + assert.equal(threadManagementBuilds, 1); + }), + ), +); diff --git a/apps/server/src/j5/a2a/runtimeLayer.ts b/apps/server/src/j5/a2a/runtimeLayer.ts new file mode 100644 index 0000000000000..a4105d9dd0cf2 --- /dev/null +++ b/apps/server/src/j5/a2a/runtimeLayer.ts @@ -0,0 +1,26 @@ +import * as Layer from "effect/Layer"; + +import { layer as deliveryWorkerLayer } from "./DeliveryWorker.ts"; +import { live as deliveryTransportLayer } from "./DeliveryTransport.ts"; +import { layer as ledgerLayer } from "./LedgerService.ts"; +import { layer as sendServiceLayer } from "./SendService.ts"; + +export const makeJ5A2ARuntimeLayer = ( + options: { + readonly ledger?: typeof ledgerLayer; + readonly deliveryTransport?: typeof deliveryTransportLayer; + } = {}, +) => { + const ledgerProvided = options.ledger ?? ledgerLayer; + const deliveryTransportProvided = options.deliveryTransport ?? deliveryTransportLayer; + const deliveryWorkerProvided = deliveryWorkerLayer.pipe( + Layer.provideMerge(deliveryTransportProvided), + ); + + return Layer.mergeAll(sendServiceLayer, deliveryWorkerProvided).pipe( + Layer.provideMerge(ledgerProvided), + ); +}; + +/** Production J5 A2A services; SQL and V2 thread management stay shared runtime dependencies. */ +export const J5A2ARuntimeLayer = makeJ5A2ARuntimeLayer(); diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 03492c3ef366e..066565db02757 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -28,6 +28,7 @@ import { import { WorktreeToolkitHandlersLive } from "./toolkits/worktree/handlers.ts"; import { WorktreeToolkit } from "./toolkits/worktree/tools.ts"; import * as WorktreeMcpService from "./WorktreeMcpService.ts"; +import { J5McpIntegrationLive } from "../j5/a2a/mcp/registration.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -243,4 +244,6 @@ export const layer = Layer.mergeAll( PreviewToolkitRegistrationLive, OrchestratorToolkitRegistrationLive, WorktreeToolkitRegistrationLive, + // J5 fork extension: one shared toolkit registration for all J5-owned tools. + J5McpIntegrationLive, ).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 3300a869fe674..5725f070fb959 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -12,6 +12,7 @@ import * as GitWorkflowService from "../../../git/GitWorkflowService.ts"; import { ThreadManagementService } from "../../../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../../../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../../../project/ProjectSetupScriptRunner.ts"; +import { SqlitePersistenceMemory } from "../../../persistence/Layers/Sqlite.ts"; import { ProviderRegistry } from "../../../provider/Services/ProviderRegistry.ts"; import { ScheduledTaskService } from "../../../scheduledTasks/ScheduledTaskService.ts"; import * as ServerSettings from "../../../serverSettings.ts"; @@ -66,6 +67,7 @@ it.effect("production mcp layer lists worktree tools over http", () => }), ), Layer.provide(PreviewAutomationBroker.layer), + Layer.provide(SqlitePersistenceMemory), Layer.provide(StubServicesLive), Layer.build, ); @@ -78,6 +80,15 @@ it.effect("production mcp layer lists worktree tools over http", () => expect(credential).toBeDefined(); const httpClient = yield* HttpClient.HttpClient; + const unauthorizedResponse = yield* httpClient.post("/mcp", { + headers: { accept: "application/json, text/event-stream" }, + body: HttpBody.text( + `{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"unauthorized","version":"1.0.0"}}}`, + "application/json", + ), + }); + expect(unauthorizedResponse.status).toBe(401); + const auth = credential!.config.authorizationHeader; const initResponse = yield* httpClient.post("/mcp", { headers: { @@ -114,6 +125,10 @@ it.effect("production mcp layer lists worktree tools over http", () => // than replacing them. expect(toolNames).toContain("preview_status"); expect(toolNames).toContain("delegate_task"); + // J5 uses the same authenticated transport and one shared registration + // that later J5 milestones extend inside the fork-owned toolkit. + expect(toolNames).toContain("send_message"); + expect(toolNames).toContain("list_participants"); // The handoff tool mutates thread state, reaches the network (origin // fetch), and runs project setup scripts, so its MCP hints must not diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 4200c6aa3d431..100b86e041fde 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -185,6 +185,54 @@ it.layer(TestLayer)("OrchestrationV2LayerLive", (it) => { }), ); + it.effect("replays an internal thread send without injecting a second message", () => + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const threadManagement = yield* ThreadManagementService; + const threadId = ThreadId.make("runtime-layer-idempotent-thread-send"); + const projectId = ProjectId.make("runtime-layer-idempotent-thread-send-project"); + const commandId = CommandId.make("runtime-layer-idempotent-thread-send-command"); + const messageId = MessageId.make("runtime-layer-idempotent-thread-send-message"); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-idempotent-thread-send-create"), + threadId, + projectId, + title: "Idempotent internal send", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-idempotent-thread-send", + }); + + const input = { + projectId, + commandId, + threadId, + messageId, + text: "Deliver exactly once", + attachments: [], + mode: "auto" as const, + createdBy: "agent" as const, + creationSource: "mcp" as const, + }; + const first = yield* threadManagement.sendToThread(input); + const replay = yield* threadManagement.sendToThread(input); + const projection = yield* threadManagement.getThreadProjection(threadId); + + assert.equal(replay.dispatch.sequence, first.dispatch.sequence); + assert.deepEqual(replay.dispatch.storedEvents, first.dispatch.storedEvents); + assert.equal(replay.message.id, first.message.id); + assert.equal(replay.run.id, first.run.id); + assert.equal(projection.messages.filter((message) => message.id === messageId).length, 1); + assert.equal(projection.runs.filter((run) => run.userMessageId === messageId).length, 1); + }), + ); + it.effect("merges an explicit provider-finished run while checkpoint capture is pending", () => Effect.gen(function* () { const orchestrator = yield* OrchestratorV2;