From dde684ccbea97abaf5a0e0e1bdc4c685f6c1a861 Mon Sep 17 00:00:00 2001 From: Lior Date: Mon, 25 May 2026 18:10:06 -0400 Subject: [PATCH 1/5] claim: task-pr-4990-nats-outbox-replay - clean replay Co-Authored-By: Codex --- .../claims/task-pr-4990-nats-outbox-replay.md | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 docs/claims/task-pr-4990-nats-outbox-replay.md diff --git a/docs/claims/task-pr-4990-nats-outbox-replay.md b/docs/claims/task-pr-4990-nats-outbox-replay.md new file mode 100644 index 0000000000..f7788529a0 --- /dev/null +++ b/docs/claims/task-pr-4990-nats-outbox-replay.md @@ -0,0 +1,20 @@ +# Claim - task-pr-4990-nats-outbox-replay + +- **Session ID:** codex/20260525T2209Z +- **Harness:** codex +- **Claimed at:** 2026-05-25T22:09:40Z +- **ETA:** 2026-05-25T22:30:00Z +- **Scope:** Replay only the intended post-#4990 agentic-organization NATS outbox and replaceable state-adapter slice onto current main, excluding stale-base governance/backlog deletions. +- **Durable target:** draft PR from a clean Codex branch after replay and focused package checks +- **Platform mirror:** https://github.com/Lucent-Financial-Group/Zeta/pull/4990 + +## Notes + +- Source branch under triage: `codex/agentic-org-package-ca-clean` at + `16417aecba9eae921bba0e363c2f5f9c4d21bbe7` as of + 2026-05-25T22:09Z. +- Do not delete or force-update that source branch during this claim. +- Current source branch diff includes stale-base noise outside scope: + `GOVERNANCE.md`, `docs/BACKLOG.md`, and a backlog row deletion. +- Dirty older #4990 local worktrees are out of scope and must remain untouched + unless a later tick explicitly proves they are safe to clean. From 248a593c5b2c5ca1f44cfe37c17548a5e791afb1 Mon Sep 17 00:00:00 2001 From: Max Chadaev Date: Mon, 25 May 2026 17:57:27 -0400 Subject: [PATCH 2/5] feat(agentic-org): add NATS outbox publisher contract Co-Authored-By: Codex --- .../docs/FIRST_IMPLEMENTATION_SLICE.md | 29 ++-- .../docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md | 25 ++- agentic-organization/packages/README.md | 18 ++- .../src/package-dependency-boundaries.test.ts | 5 + .../src/package-dependency-boundaries.ts | 1 + .../packages/messaging-nats/src/index.ts | 7 + .../nats-jetstream-event-publisher.test.ts | 103 +++++++++++++ .../src/nats-jetstream-event-publisher.ts | 49 ++++++ .../packages/messaging/src/index.ts | 15 ++ .../messaging/src/outbox-publisher.test.ts | 145 ++++++++++++++++++ .../messaging/src/outbox-publisher.ts | 114 ++++++++++++++ .../src/cockroach-outbox-event-source.test.ts | 107 +++++++++++++ .../src/cockroach-outbox-event-source.ts | 75 +++++++++ .../packages/state-cockroach/src/index.ts | 8 + openspec/specs/agentic-organization/spec.md | 23 +++ 15 files changed, 703 insertions(+), 21 deletions(-) create mode 100644 agentic-organization/packages/messaging-nats/src/index.ts create mode 100644 agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts create mode 100644 agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.ts create mode 100644 agentic-organization/packages/messaging/src/outbox-publisher.test.ts create mode 100644 agentic-organization/packages/messaging/src/outbox-publisher.ts create mode 100644 agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts create mode 100644 agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts diff --git a/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md b/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md index 69ae71783e..8e43283cf0 100644 --- a/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md +++ b/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md @@ -9,10 +9,10 @@ Implemented as a small NodeNext TypeScript package slice. This slice turns the first Agentic Organization runtime contract from architecture prose into executable TypeScript. -It does not introduce NestJS, CockroachDB, NATS clients, Temporal, -Dapr, Hermes, Hindsight, or Kubernetes deployment manifests yet. Those -remain adapter layers. The goal is to prove the Organization command -shape before adding distributed infrastructure. +It does not introduce NestJS, live NATS connections, Temporal, Dapr, +Hermes, Hindsight, or Kubernetes deployment manifests yet. Those remain +adapter layers. The goal is to prove the Organization command and event +publication shape before adding distributed infrastructure. The slice is intentionally generic. `send_supervisor_signal` is the coordination primitive; specific downstream outcomes are lifecycle @@ -29,6 +29,8 @@ send_supervisor_signal -> chain-of-command signal -> audit event -> outbox event with canonical event envelope + -> outbox publisher + -> NATS JetStream event publisher adapter -> NATS subject contract -> LGTM span attributes -> supervisor triage reaction plan @@ -41,8 +43,9 @@ send_supervisor_signal | `@agentic-org/domain` | event envelope, command/event constants, aggregate constants, supervisor-chain communication types, hat communication briefs, work item state machine, shared records | | `@agentic-org/application` | command pipeline, command-handler registry, state-store ports, idempotency conflict handling, supervisor signal handler | | `@agentic-org/state` | in-memory Organization state-store factory fake | -| `@agentic-org/state-cockroach` | CockroachDB state-store factory contract, SQL statement catalog, and first core-state migration skeleton | -| `@agentic-org/messaging` | stable `agentic-org....` subject builder | +| `@agentic-org/state-cockroach` | CockroachDB state-store/outbox-source contracts, SQL statement catalogs, and first core-state migration skeleton | +| `@agentic-org/messaging` | stable `agentic-org....` subject builder, outbox publisher, event publisher port, and typed domain resolver | +| `@agentic-org/messaging-nats` | NATS JetStream event publisher adapter contract with canonical JSON payloads, headers, and message IDs | | `@agentic-org/observability` | OpenTelemetry/LGTM span attribute projection | | `@agentic-org/runtime` | first rule that plans triage for the target supervisor when a chain signal is sent | | `@agentic-org/governance` | package dependency-boundary checks that prevent application code from importing concrete state/runtime adapters | @@ -102,6 +105,11 @@ Hermes runs, MCP calls, and UI evidence. - A governance test enforces that application code does not import the state adapter, Cockroach adapter, NestJS, NATS, Dapr, Temporal, Drizzle, or Postgres clients. +- The outbox publisher claims unpublished events, publishes each event + through an `EventPublisher` port, and marks rows published only after + the publish succeeds. +- The NATS adapter publishes canonical JSON envelopes with typed headers + and event IDs as message IDs for idempotent JetStream publication. - Duplicate commands with the same idempotency key and request hash replay the stored result. - Duplicate commands with the same idempotency key and a different @@ -113,11 +121,10 @@ Hermes runs, MCP calls, and UI evidence. ## Next Slice -The next slice should turn the CockroachDB adapter contract into a -transactional integration test once a local/dev Cockroach connection is -available, then add the NATS outbox publisher worker. The worker can -publish persisted outbox rows to JetStream and attach the same telemetry -attributes. +The next slice should add inbox/consumer dedupe before automation starts +performing side effects from NATS events. After that, wire the outbox +publisher into a worker host and add a transactional Cockroach +integration test once a local/dev Cockroach connection is available. Do not make the next slice a pile of bespoke request commands. Build the generic supervisor triage lifecycle first, then let specialized diff --git a/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md b/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md index 2479f6bc94..23b382cc3c 100644 --- a/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md +++ b/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md @@ -164,8 +164,9 @@ calling them. | Package | Owns | | ---------------------------------------- | ----------------------------------------------------------------------------------------------- | | `@agentic-org/state` | Drizzle schema, migrations, repositories, transactions, outbox, inbox, idempotency, leases | -| `@agentic-org/state-cockroach` | CockroachDB implementation of state-store ports, SQL statement catalog, and migration contracts | +| `@agentic-org/state-cockroach` | CockroachDB implementation of state-store and outbox-source ports, SQL catalogs, migrations | | `@agentic-org/messaging` | NATS envelope builder, subject builder, JetStream publisher, consumer, DLQ, replay contracts | +| `@agentic-org/messaging-nats` | NATS JetStream implementation of the event publisher port, canonical JSON, headers, message IDs | | `@agentic-org/workflows-temporal` | Temporal workflow and activity contracts, task queues, workflow clients | | `@agentic-org/actors-dapr` | Dapr actor interfaces, actor implementations, reminders, actor state projection | | `@agentic-org/mcp` | MCP schemas, tool registry, preflight checks, policy-checked tool handlers | @@ -344,6 +345,15 @@ type AgenticEventEnvelope = { No app should publish raw NATS payloads directly. Publishing should go through `@agentic-org/messaging`. +The generic outbox publisher should claim unpublished outbox events from +an `OutboxEventSource`, resolve the typed Organization messaging +domain, publish through an `EventPublisher` port, and mark the outbox +row published only after the publish succeeds. The NATS adapter is an +implementation of that port; it owns transport-specific concerns such as +headers, message IDs, and JSON serialization. This keeps the +Organization event loop extensible and testable without coupling the +publisher to the NATS client. + ### Event-to-Automation Contract Agentic Organization should behave like an event-driven operating system. @@ -761,6 +771,7 @@ other work. - `@agentic-org/state-cockroach`; - `@agentic-org/policy`; - `@agentic-org/messaging`; + - `@agentic-org/messaging-nats`; - `@agentic-org/observability`; - `@agentic-org/work-os`; - `@agentic-org/hats`; @@ -786,13 +797,15 @@ other work. 5. Use fake adapters for Hermes, Hindsight, Dapr, Temporal, and hat-system. 6. Add NATS outbox publisher and one consumer after command tests pass. -7. Add the first rule catalog and reaction executor for ready work, +7. Add inbox/consumer dedupe before any NATS-driven automation performs + side effects. +8. Add the first rule catalog and reaction executor for ready work, review staffing, QA staffing, blocker escalation, and late run incidents. -8. Add the NestJS API and worker hosts. -9. Add UI projections for work board, review center, and evidence - timeline. -10. Add real cluster adapters one at a time. +9. Add the NestJS API and worker hosts. +10. Add UI projections for work board, review center, and evidence + timeline. +11. Add real cluster adapters one at a time. ## Extraction Path diff --git a/agentic-organization/packages/README.md b/agentic-organization/packages/README.md index 37a4f34893..fd0bbea7dc 100644 --- a/agentic-organization/packages/README.md +++ b/agentic-organization/packages/README.md @@ -2,8 +2,8 @@ These packages are the first executable slice of Agentic Organization. They are intentionally small and run as a NodeNext TypeScript island -before any NestJS host, CockroachDB adapter, NATS client, Temporal -worker, Dapr actor host, or Kubernetes deployment is introduced. +before any NestJS host, live NATS connection, Temporal worker, Dapr actor +host, or Kubernetes deployment is introduced. ## Package Boundary @@ -12,8 +12,9 @@ worker, Dapr actor host, or Kubernetes deployment is introduced. | `domain` | typed command names, event names, aggregate names, work item state machine, event envelope, shared records | | `application` | command pipeline, handler registry, idempotency handling, state-store ports, first supervisor-chain signal command handler | | `state` | in-memory Organization state-store factory used as the first repository port fake | -| `state-cockroach` | CockroachDB state-store factory contract, SQL statement catalog, and first core-state migration skeleton | -| `messaging` | NATS subject contract without a live NATS dependency | +| `state-cockroach` | CockroachDB state-store/outbox-source contracts, SQL statement catalogs, and first core-state migration skeleton | +| `messaging` | NATS subject contract, outbox publisher port, event publisher port, and domain resolver without a live NATS dependency | +| `messaging-nats` | NATS JetStream event publisher adapter contract with canonical JSON payloads, headers, and message IDs | | `observability` | LGTM/OpenTelemetry attribute projection from Agentic event envelopes | | `runtime` | first event-to-automation reaction rule | | `governance` | package dependency-boundary checks that keep core packages SOLID and adapter-free | @@ -28,6 +29,8 @@ supervisor-chain signal command -> chain communication record -> audit event -> outbox event + -> outbox publisher + -> NATS JetStream event publisher adapter -> NATS subject / telemetry contract -> automation reaction plan ``` @@ -47,6 +50,13 @@ central `switch` or `if` dispatcher. real CockroachDB adapter must not be squeezed into a synchronous toy shape. +The outbox publisher owns the generic publish loop: claim unpublished +outbox rows, resolve a typed Organization domain, publish through an +`EventPublisher` port, and mark the outbox row published only after the +publish returns successfully. The NATS package implements that port and +is the only package in this slice that knows about NATS headers, +message IDs, and JSON transport payloads. + ## Validation Run the package tests from `agentic-organization/`: diff --git a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts index c954529300..b9ae9bdd1e 100644 --- a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts +++ b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts @@ -25,6 +25,11 @@ describe("package dependency boundaries", () => { "postgres", ], }, + { + packageName: PackageBoundaryRule.Messaging, + sourceGlob: "messaging/src/**/*.ts", + forbiddenImportFragments: ["../messaging-nats", "../../messaging-nats", "nats"], + }, ], }); diff --git a/agentic-organization/packages/governance/src/package-dependency-boundaries.ts b/agentic-organization/packages/governance/src/package-dependency-boundaries.ts index 01df149e12..08bfb7386a 100644 --- a/agentic-organization/packages/governance/src/package-dependency-boundaries.ts +++ b/agentic-organization/packages/governance/src/package-dependency-boundaries.ts @@ -4,6 +4,7 @@ import { fileURLToPath } from "node:url"; export const PackageBoundaryRule = { Application: "application", + Messaging: "messaging", } as const; export type PackageBoundaryRule = (typeof PackageBoundaryRule)[keyof typeof PackageBoundaryRule]; diff --git a/agentic-organization/packages/messaging-nats/src/index.ts b/agentic-organization/packages/messaging-nats/src/index.ts new file mode 100644 index 0000000000..9bc7e8466b --- /dev/null +++ b/agentic-organization/packages/messaging-nats/src/index.ts @@ -0,0 +1,7 @@ +export { + NatsHeaderName, + createNatsJetStreamEventPublisher, + type CreateNatsJetStreamEventPublisherInput, + type NatsJetStreamClient, + type NatsJetStreamMessage, +} from "./nats-jetstream-event-publisher.ts"; diff --git a/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts new file mode 100644 index 0000000000..b565bdc65f --- /dev/null +++ b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts @@ -0,0 +1,103 @@ +import { deepEqual, equal } from "node:assert/strict"; +import { describe, test } from "node:test"; + +import { + AgenticAggregateType, + AgenticEventType, + createAgenticEventEnvelope, + type OutboxEvent, +} from "../../domain/src/index.ts"; +import { + createNatsJetStreamEventPublisher, + NatsHeaderName, + type NatsJetStreamClient, +} from "./nats-jetstream-event-publisher.ts"; + +describe("NATS JetStream event publisher", () => { + test("publishes canonical JSON with idempotent headers and message ID", async () => { + const client = createRecordingNatsClient(); + const publisher = createNatsJetStreamEventPublisher({ + client, + }); + const outboxEvent = createOutboxEvent(); + + await publisher.publish({ + subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + outboxEvent, + }); + + equal(client.messages.length, 1); + deepEqual(client.messages[0], { + subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + payload: JSON.stringify(outboxEvent.envelope), + messageId: "evt-001", + headers: { + [NatsHeaderName.EventId]: "evt-001", + [NatsHeaderName.EventType]: AgenticEventType.SupervisorSignalSent, + [NatsHeaderName.CorrelationId]: "corr-001", + [NatsHeaderName.CausationId]: "cause-001", + [NatsHeaderName.TraceId]: "trace-001", + [NatsHeaderName.IdempotencyKey]: "idem-001", + [NatsHeaderName.OutboxEventId]: "outbox-001", + }, + }); + }); +}); + +function createRecordingNatsClient(): NatsJetStreamClient & { + messages: { + subject: string; + payload: string; + messageId: string; + headers: Record; + }[]; +} { + const messages: { + subject: string; + payload: string; + messageId: string; + headers: Record; + }[] = []; + + return { + messages, + publish: async (message) => { + messages.push(message); + }, + }; +} + +function createOutboxEvent(): OutboxEvent { + return { + outboxEventId: "outbox-001", + envelope: createAgenticEventEnvelope({ + eventId: "evt-001", + eventType: AgenticEventType.SupervisorSignalSent, + occurredAt: "2026-05-25T20:00:00.000Z", + actor: { + agentId: "agent-developer-001", + hatAssignmentId: "hat-assignment-dev-001", + }, + scope: { + organizationId: "org-lfg", + projectId: "project-agentic-org", + workItemId: "work-outbox-001", + }, + aggregate: { + aggregateId: "supervisor-signal-001", + aggregateType: AgenticAggregateType.SupervisorSignal, + aggregateVersion: 1, + }, + trace: { + commandId: "cmd-001", + correlationId: "corr-001", + causationId: "cause-001", + traceId: "trace-001", + idempotencyKey: "idem-001", + }, + payload: { + title: "Blocked on scoped NATS publisher", + }, + }), + }; +} diff --git a/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.ts b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.ts new file mode 100644 index 0000000000..ec5a24b518 --- /dev/null +++ b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.ts @@ -0,0 +1,49 @@ +import type { EventPublisher } from "../../messaging/src/index.ts"; + +export const NatsHeaderName = { + EventId: "Nats-Msg-Event-Id", + EventType: "Nats-Msg-Event-Type", + CorrelationId: "Nats-Msg-Correlation-Id", + CausationId: "Nats-Msg-Causation-Id", + TraceId: "Nats-Msg-Trace-Id", + IdempotencyKey: "Nats-Msg-Idempotency-Key", + OutboxEventId: "Nats-Msg-Outbox-Event-Id", +} as const; + +export type NatsHeaderName = (typeof NatsHeaderName)[keyof typeof NatsHeaderName]; + +export type NatsJetStreamMessage = { + subject: string; + payload: string; + messageId: string; + headers: Record; +}; + +export type NatsJetStreamClient = { + publish: (message: NatsJetStreamMessage) => Promise; +}; + +export type CreateNatsJetStreamEventPublisherInput = { + client: NatsJetStreamClient; +}; + +export function createNatsJetStreamEventPublisher(input: CreateNatsJetStreamEventPublisherInput): EventPublisher { + return { + publish: async (publication) => { + await input.client.publish({ + subject: publication.subject, + payload: JSON.stringify(publication.outboxEvent.envelope), + messageId: publication.outboxEvent.envelope.eventId, + headers: { + [NatsHeaderName.EventId]: publication.outboxEvent.envelope.eventId, + [NatsHeaderName.EventType]: publication.outboxEvent.envelope.eventType, + [NatsHeaderName.CorrelationId]: publication.outboxEvent.envelope.trace.correlationId, + [NatsHeaderName.CausationId]: publication.outboxEvent.envelope.trace.causationId, + [NatsHeaderName.TraceId]: publication.outboxEvent.envelope.trace.traceId, + [NatsHeaderName.IdempotencyKey]: publication.outboxEvent.envelope.trace.idempotencyKey, + [NatsHeaderName.OutboxEventId]: publication.outboxEvent.outboxEventId, + }, + }); + }, + }; +} diff --git a/agentic-organization/packages/messaging/src/index.ts b/agentic-organization/packages/messaging/src/index.ts index 134ca99eb0..2cb19ab88e 100644 --- a/agentic-organization/packages/messaging/src/index.ts +++ b/agentic-organization/packages/messaging/src/index.ts @@ -1 +1,16 @@ export { AgenticSubjectPrefix, buildAgenticEventSubject, type AgenticEventSubjectInput } from "./subject-builder.ts"; +export { + AgenticMessagingDomain, + OutboxPublishOutcomeStatus, + createOutboxPublisher, + resolveAgenticMessagingDomain, + type ClaimUnpublishedOutboxEventsInput, + type CreateOutboxPublisherInput, + type EventPublication, + type EventPublisher, + type MarkOutboxEventPublishedInput, + type OutboxEventSource, + type OutboxPublishBatchResult, + type OutboxPublisher, + type ResolveAgenticMessagingDomain, +} from "./outbox-publisher.ts"; diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.test.ts b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts new file mode 100644 index 0000000000..4de151fdc3 --- /dev/null +++ b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts @@ -0,0 +1,145 @@ +import { deepEqual, equal } from "node:assert/strict"; +import { describe, test } from "node:test"; + +import { + AgenticAggregateType, + AgenticEventType, + createAgenticEventEnvelope, + type OutboxEvent, +} from "../../domain/src/index.ts"; +import { + AgenticMessagingDomain, + OutboxPublishOutcomeStatus, + createOutboxPublisher, + resolveAgenticMessagingDomain, + type EventPublication, + type EventPublisher, + type OutboxEventSource, +} from "./outbox-publisher.ts"; + +describe("outbox publisher", () => { + test("resolves event domains through typed mappings", () => { + deepEqual( + resolveAgenticMessagingDomain(AgenticEventType.SupervisorSignalSent), + AgenticMessagingDomain.SupervisorSignal, + ); + }); + + test("publishes unpublished outbox events and marks them published", async () => { + const outboxEvent = createOutboxEvent(); + const outboxSource = createRecordingOutboxSource([outboxEvent]); + const eventPublisher = createRecordingEventPublisher(); + const publisher = createOutboxPublisher({ + outboxSource, + eventPublisher, + environment: "local", + resolveDomain: resolveAgenticMessagingDomain, + now: () => "2026-05-25T21:00:00.000Z", + }); + + const result = await publisher.publishNextBatch({ + batchSize: 10, + }); + + deepEqual(result, { + status: OutboxPublishOutcomeStatus.Published, + attemptedCount: 1, + publishedOutboxEventIds: ["outbox-001"], + }); + deepEqual(outboxSource.markedPublished, [ + { + outboxEventId: "outbox-001", + publishedAt: "2026-05-25T21:00:00.000Z", + }, + ]); + deepEqual(eventPublisher.publications, [ + { + subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + outboxEvent, + }, + ]); + }); + + test("returns empty when there is no work to publish", async () => { + const outboxSource = createRecordingOutboxSource([]); + const eventPublisher = createRecordingEventPublisher(); + const publisher = createOutboxPublisher({ + outboxSource, + eventPublisher, + environment: "local", + resolveDomain: resolveAgenticMessagingDomain, + now: () => "2026-05-25T21:00:00.000Z", + }); + + const result = await publisher.publishNextBatch({ + batchSize: 10, + }); + + equal(result.status, OutboxPublishOutcomeStatus.Empty); + equal(eventPublisher.publications.length, 0); + equal(outboxSource.markedPublished.length, 0); + }); +}); + +function createRecordingOutboxSource(outboxEvents: OutboxEvent[]): OutboxEventSource & { + markedPublished: { outboxEventId: string; publishedAt: string }[]; +} { + const markedPublished: { outboxEventId: string; publishedAt: string }[] = []; + + return { + markedPublished, + claimUnpublishedOutboxEvents: async () => outboxEvents, + markOutboxEventPublished: async (input) => { + markedPublished.push(input); + }, + }; +} + +function createRecordingEventPublisher(): EventPublisher & { + publications: EventPublication[]; +} { + const publications: EventPublication[] = []; + + return { + publications, + publish: async (publication) => { + publications.push(publication); + }, + }; +} + +function createOutboxEvent(): OutboxEvent { + return { + outboxEventId: "outbox-001", + envelope: createAgenticEventEnvelope({ + eventId: "evt-001", + eventType: AgenticEventType.SupervisorSignalSent, + occurredAt: "2026-05-25T20:00:00.000Z", + actor: { + agentId: "agent-developer-001", + hatAssignmentId: "hat-assignment-dev-001", + }, + scope: { + organizationId: "org-lfg", + projectId: "project-agentic-org", + teamId: "team-runtime", + workItemId: "work-outbox-001", + }, + aggregate: { + aggregateId: "supervisor-signal-001", + aggregateType: AgenticAggregateType.SupervisorSignal, + aggregateVersion: 1, + }, + trace: { + commandId: "cmd-001", + correlationId: "corr-001", + causationId: "cause-001", + traceId: "trace-001", + idempotencyKey: "idem-001", + }, + payload: { + title: "Blocked on scoped NATS publisher", + }, + }), + }; +} diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.ts b/agentic-organization/packages/messaging/src/outbox-publisher.ts new file mode 100644 index 0000000000..0897a35274 --- /dev/null +++ b/agentic-organization/packages/messaging/src/outbox-publisher.ts @@ -0,0 +1,114 @@ +import { AgenticEventType, type OutboxEvent } from "../../domain/src/index.ts"; +import { buildAgenticEventSubject } from "./subject-builder.ts"; + +export const AgenticMessagingDomain = { + SupervisorSignal: "supervisor_signal", + WorkItem: "work_item", +} as const; + +export type AgenticMessagingDomain = (typeof AgenticMessagingDomain)[keyof typeof AgenticMessagingDomain]; + +export const OutboxPublishOutcomeStatus = { + Empty: "empty", + Published: "published", +} as const; + +export type OutboxPublishOutcomeStatus = (typeof OutboxPublishOutcomeStatus)[keyof typeof OutboxPublishOutcomeStatus]; + +export type ClaimUnpublishedOutboxEventsInput = { + batchSize: number; +}; + +export type MarkOutboxEventPublishedInput = { + outboxEventId: string; + publishedAt: string; +}; + +export type OutboxEventSource = { + claimUnpublishedOutboxEvents: (input: ClaimUnpublishedOutboxEventsInput) => Promise; + markOutboxEventPublished: (input: MarkOutboxEventPublishedInput) => Promise; +}; + +export type EventPublication = { + subject: string; + outboxEvent: OutboxEvent; +}; + +export type EventPublisher = { + publish: (publication: EventPublication) => Promise; +}; + +export type ResolveAgenticMessagingDomain = (eventType: AgenticEventType) => AgenticMessagingDomain; + +export type OutboxPublisher = { + publishNextBatch: (input: ClaimUnpublishedOutboxEventsInput) => Promise; +}; + +export type OutboxPublishBatchResult = { + status: OutboxPublishOutcomeStatus; + attemptedCount: number; + publishedOutboxEventIds: string[]; +}; + +export type CreateOutboxPublisherInput = { + outboxSource: OutboxEventSource; + eventPublisher: EventPublisher; + environment: string; + resolveDomain: ResolveAgenticMessagingDomain; + now: () => string; +}; + +export function createOutboxPublisher(input: CreateOutboxPublisherInput): OutboxPublisher { + return { + publishNextBatch: async (publishInput) => { + const outboxEvents = await input.outboxSource.claimUnpublishedOutboxEvents(publishInput); + + if (outboxEvents.length === 0) { + return { + status: OutboxPublishOutcomeStatus.Empty, + attemptedCount: 0, + publishedOutboxEventIds: [], + }; + } + + const publishedOutboxEventIds: string[] = []; + + for (const outboxEvent of outboxEvents) { + const subject = buildAgenticEventSubject({ + environment: input.environment, + organizationId: outboxEvent.envelope.scope.organizationId, + domain: input.resolveDomain(outboxEvent.envelope.eventType), + eventType: outboxEvent.envelope.eventType, + }); + + await input.eventPublisher.publish({ + subject, + outboxEvent, + }); + await input.outboxSource.markOutboxEventPublished({ + outboxEventId: outboxEvent.outboxEventId, + publishedAt: input.now(), + }); + publishedOutboxEventIds.push(outboxEvent.outboxEventId); + } + + return { + status: OutboxPublishOutcomeStatus.Published, + attemptedCount: outboxEvents.length, + publishedOutboxEventIds, + }; + }, + }; +} + +export function resolveAgenticMessagingDomain(eventType: AgenticEventType): AgenticMessagingDomain { + if (eventType === AgenticEventType.SupervisorSignalSent) { + return AgenticMessagingDomain.SupervisorSignal; + } + + if (eventType === AgenticEventType.WorkItemChanged || eventType === AgenticEventType.WorkItemStateChanged) { + return AgenticMessagingDomain.WorkItem; + } + + throw new Error(`unsupported event type for messaging domain: ${eventType}`); +} diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts new file mode 100644 index 0000000000..a5e6eda061 --- /dev/null +++ b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts @@ -0,0 +1,107 @@ +import { deepEqual } from "node:assert/strict"; +import { describe, test } from "node:test"; + +import { AgenticAggregateType, AgenticEventType, type AgenticEventEnvelope } from "../../domain/src/index.ts"; +import { + CockroachOutboxEventSourceStatement, + createCockroachOutboxEventSource, + type CockroachOutboxSqlExecutor, + type CockroachOutboxSqlStatement, +} from "./cockroach-outbox-event-source.ts"; + +describe("cockroach outbox event source", () => { + test("claims unpublished outbox events and marks them published", async () => { + const executor = createRecordingExecutor(); + const outboxSource = createCockroachOutboxEventSource({ + executor, + }); + + const outboxEvents = await outboxSource.claimUnpublishedOutboxEvents({ + batchSize: 10, + }); + await outboxSource.markOutboxEventPublished({ + outboxEventId: "outbox-001", + publishedAt: "2026-05-25T21:00:00.000Z", + }); + + deepEqual(outboxEvents, [ + { + outboxEventId: "outbox-001", + envelope: createEnvelope(), + }, + ]); + deepEqual( + executor.statements.map((statement) => statement.name), + [ + CockroachOutboxEventSourceStatement.ClaimUnpublishedOutboxEvents, + CockroachOutboxEventSourceStatement.MarkOutboxEventPublished, + ], + ); + }); +}); + +type RecordingCockroachOutboxSqlExecutor = CockroachOutboxSqlExecutor & { + statements: { name: CockroachOutboxEventSourceStatement; parameters: readonly unknown[] }[]; +}; + +function createRecordingExecutor(): RecordingCockroachOutboxSqlExecutor { + const statements: { name: CockroachOutboxEventSourceStatement; parameters: readonly unknown[] }[] = []; + + return { + statements, + execute: async >(statement: CockroachOutboxSqlStatement) => { + statements.push(statement); + + if (statement.name === CockroachOutboxEventSourceStatement.ClaimUnpublishedOutboxEvents) { + return { + rows: [ + { + outbox_event_id: "outbox-001", + envelope_json: createEnvelope(), + }, + ] as Row[], + }; + } + + return { + rows: [], + }; + }, + }; +} + +function createEnvelope(): AgenticEventEnvelope { + return { + eventId: "evt-001", + eventType: AgenticEventType.SupervisorSignalSent, + schemaVersion: "agentic.org.event.v1", + occurredAt: "2026-05-25T20:00:00.000Z", + actor: { + agentId: "agent-developer-001", + hatAssignmentId: "hat-assignment-dev-001", + }, + scope: { + organizationId: "org-lfg", + projectId: "project-agentic-org", + workItemId: "work-outbox-001", + }, + aggregate: { + aggregateId: "supervisor-signal-001", + aggregateType: AgenticAggregateType.SupervisorSignal, + aggregateVersion: 1, + }, + trace: { + commandId: "cmd-001", + correlationId: "corr-001", + causationId: "cause-001", + traceId: "trace-001", + idempotencyKey: "idem-001", + }, + replay: { + isReplay: false, + }, + payload: { + title: "Blocked on scoped NATS publisher", + }, + }; +} diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts new file mode 100644 index 0000000000..28b74733ba --- /dev/null +++ b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts @@ -0,0 +1,75 @@ +import type { AgenticEventEnvelope } from "../../domain/src/index.ts"; +import type { OutboxEventSource } from "../../messaging/src/index.ts"; +import { CockroachTableName } from "./cockroach-schema.ts"; + +export const CockroachOutboxEventSourceStatement = { + ClaimUnpublishedOutboxEvents: "claim_unpublished_outbox_events", + MarkOutboxEventPublished: "mark_outbox_event_published", +} as const; + +export type CockroachOutboxEventSourceStatement = + (typeof CockroachOutboxEventSourceStatement)[keyof typeof CockroachOutboxEventSourceStatement]; + +export type CockroachOutboxSqlStatement = { + name: CockroachOutboxEventSourceStatement; + sql: string; + parameters: readonly unknown[]; +}; + +export type CockroachOutboxSqlResult> = { + rows: readonly Row[]; +}; + +export type CockroachOutboxSqlExecutor = { + execute: >( + statement: CockroachOutboxSqlStatement, + ) => Promise>; +}; + +export type CreateCockroachOutboxEventSourceInput = { + executor: CockroachOutboxSqlExecutor; +}; + +export function createCockroachOutboxEventSource(input: CreateCockroachOutboxEventSourceInput): OutboxEventSource { + return { + claimUnpublishedOutboxEvents: async (claimInput) => { + const result = await input.executor.execute({ + name: CockroachOutboxEventSourceStatement.ClaimUnpublishedOutboxEvents, + sql: CockroachOutboxEventSourceSql.ClaimUnpublishedOutboxEvents, + parameters: [claimInput.batchSize], + }); + + return result.rows.map((row) => ({ + outboxEventId: row.outbox_event_id, + envelope: row.envelope_json, + })); + }, + markOutboxEventPublished: async (markInput) => { + await input.executor.execute({ + name: CockroachOutboxEventSourceStatement.MarkOutboxEventPublished, + sql: CockroachOutboxEventSourceSql.MarkOutboxEventPublished, + parameters: [markInput.outboxEventId, markInput.publishedAt], + }); + }, + }; +} + +type OutboxEventRow = { + outbox_event_id: string; + envelope_json: AgenticEventEnvelope; +}; + +const CockroachOutboxEventSourceSql = { + ClaimUnpublishedOutboxEvents: ` + SELECT outbox_event_id, envelope_json + FROM ${CockroachTableName.OutboxEvents} + WHERE published_at IS NULL + ORDER BY outbox_event_id + LIMIT $1 + `, + MarkOutboxEventPublished: ` + UPDATE ${CockroachTableName.OutboxEvents} + SET published_at = $2 + WHERE outbox_event_id = $1 + `, +} as const; diff --git a/agentic-organization/packages/state-cockroach/src/index.ts b/agentic-organization/packages/state-cockroach/src/index.ts index c98453bb3a..bf795c6d1b 100644 --- a/agentic-organization/packages/state-cockroach/src/index.ts +++ b/agentic-organization/packages/state-cockroach/src/index.ts @@ -6,6 +6,14 @@ export { type CockroachSqlStatement, type CreateCockroachCommandStateStoreFactoryInput, } from "./cockroach-command-state-store.ts"; +export { + CockroachOutboxEventSourceStatement, + createCockroachOutboxEventSource, + type CockroachOutboxSqlExecutor, + type CockroachOutboxSqlResult, + type CockroachOutboxSqlStatement, + type CreateCockroachOutboxEventSourceInput, +} from "./cockroach-outbox-event-source.ts"; export { CockroachCoreStateMigrationName, CockroachTableName, diff --git a/openspec/specs/agentic-organization/spec.md b/openspec/specs/agentic-organization/spec.md index 605dba6a45..71f30853f1 100644 --- a/openspec/specs/agentic-organization/spec.md +++ b/openspec/specs/agentic-organization/spec.md @@ -142,6 +142,29 @@ Organization NATS subjects MUST use a stable organization-scoped shape. - **THEN** the subject shape is `agentic-org....` +### Requirement: Outbox publisher is idempotent and adapter-backed + +Organization outbox publication MUST be driven by a generic publisher +and a concrete event-publisher adapter. + +#### Scenario: Outbox event is published + +- **WHEN** unpublished outbox events are claimed +- **THEN** the publisher resolves the typed Organization messaging + domain and builds the stable NATS subject +- **AND** the publisher sends the event through an `EventPublisher` port +- **AND** the outbox row is marked published only after the publish + succeeds + +#### Scenario: NATS adapter publishes event + +- **WHEN** the NATS JetStream adapter publishes an event publication +- **THEN** it sends the canonical event envelope as JSON +- **AND** it uses the event ID as the message ID +- **AND** it includes typed headers for event ID, event type, + correlation ID, causation ID, trace ID, idempotency key, and outbox + event ID + ### Requirement: Telemetry is complete at the event boundary Organization packages MUST expose OpenTelemetry-compatible attributes From 855add85479083db57eddfdcc12f5c39f86c9f0d Mon Sep 17 00:00:00 2001 From: Max Chadaev Date: Mon, 25 May 2026 18:06:57 -0400 Subject: [PATCH 3/5] refactor(agentic-org): keep durable state adapter replaceable Co-Authored-By: Codex --- .../docs/FIRST_IMPLEMENTATION_SLICE.md | 18 +- .../docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md | 44 ++-- .../docs/V0_EXECUTABLE_CONTRACT.md | 51 ++--- .../docs/V0_SCHEMA_AND_COMMANDS.md | 210 +++++++++--------- agentic-organization/packages/README.md | 26 ++- .../src/package-dependency-boundaries.test.ts | 5 + .../src/package-dependency-boundaries.ts | 1 + .../packages/messaging/src/index.ts | 3 - .../messaging/src/outbox-publisher.test.ts | 2 +- .../messaging/src/outbox-publisher.ts | 15 +- .../src/cockroach-outbox-event-source.ts | 2 +- .../packages/state/src/index.ts | 5 + .../packages/state/src/outbox-event-source.ts | 15 ++ openspec/specs/agentic-organization/spec.md | 15 +- 14 files changed, 225 insertions(+), 187 deletions(-) create mode 100644 agentic-organization/packages/state/src/outbox-event-source.ts diff --git a/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md b/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md index 8e43283cf0..9f8e615eba 100644 --- a/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md +++ b/agentic-organization/docs/FIRST_IMPLEMENTATION_SLICE.md @@ -42,8 +42,8 @@ send_supervisor_signal | ------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `@agentic-org/domain` | event envelope, command/event constants, aggregate constants, supervisor-chain communication types, hat communication briefs, work item state machine, shared records | | `@agentic-org/application` | command pipeline, command-handler registry, state-store ports, idempotency conflict handling, supervisor signal handler | -| `@agentic-org/state` | in-memory Organization state-store factory fake | -| `@agentic-org/state-cockroach` | CockroachDB state-store/outbox-source contracts, SQL statement catalogs, and first core-state migration skeleton | +| `@agentic-org/state` | generic state-store/outbox-source ports plus the in-memory Organization state-store factory fake | +| `@agentic-org/state-cockroach` | first replaceable durable SQL implementation of the state-store/outbox-source ports, backed by CockroachDB | | `@agentic-org/messaging` | stable `agentic-org....` subject builder, outbox publisher, event publisher port, and typed domain resolver | | `@agentic-org/messaging-nats` | NATS JetStream event publisher adapter contract with canonical JSON payloads, headers, and message IDs | | `@agentic-org/observability` | OpenTelemetry/LGTM span attribute projection | @@ -99,12 +99,15 @@ Hermes runs, MCP calls, and UI evidence. - The command pipeline receives state-store factories and command handlers through ports instead of constructing in-memory adapters or branching on command types. -- State-store ports are async from the beginning so CockroachDB, - NATS-backed workers, and other real adapters do not inherit a fake - synchronous shape. +- State-store and outbox-source ports are async from the beginning so + durable SQL, NATS-backed workers, and other real adapters do not + inherit a fake synchronous shape. - A governance test enforces that application code does not import the state adapter, Cockroach adapter, NestJS, NATS, Dapr, Temporal, Drizzle, or Postgres clients. +- A governance test enforces that the Cockroach state adapter does not + import messaging, NATS, or JetStream. Durable state can be swapped + without dragging transport concerns into the repository layer. - The outbox publisher claims unpublished events, publishes each event through an `EventPublisher` port, and marks rows published only after the publish succeeds. @@ -123,8 +126,9 @@ Hermes runs, MCP calls, and UI evidence. The next slice should add inbox/consumer dedupe before automation starts performing side effects from NATS events. After that, wire the outbox -publisher into a worker host and add a transactional Cockroach -integration test once a local/dev Cockroach connection is available. +publisher into a worker host and add a transactional durable-state +adapter integration test using CockroachDB as the first cluster-backed +implementation once a local/dev connection is available. Do not make the next slice a pile of bespoke request commands. Build the generic supervisor triage lifecycle first, then let specialized diff --git a/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md b/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md index 23b382cc3c..5462e2b839 100644 --- a/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md +++ b/agentic-organization/docs/TECHNICAL_CA_PACKAGE_ARCHITECTURE.md @@ -55,7 +55,8 @@ The Organization owns: The cluster provides: -- CockroachDB for authoritative Organization state; +- CockroachDB as the first durable SQL adapter for authoritative + Organization state; - NATS JetStream for event transport, fanout, inboxes, replay, and DLQ; - Temporal TS for durable long-running workflows; - Dapr Actors for hot entity-local coordination; @@ -77,7 +78,7 @@ Runtime host -> command handler registry -> policy check -> domain state transition - -> CockroachDB transaction + -> durable state transaction through the state adapter -> authoritative state -> audit event -> outbox event @@ -163,8 +164,8 @@ calling them. | Package | Owns | | ---------------------------------------- | ----------------------------------------------------------------------------------------------- | -| `@agentic-org/state` | Drizzle schema, migrations, repositories, transactions, outbox, inbox, idempotency, leases | -| `@agentic-org/state-cockroach` | CockroachDB implementation of state-store and outbox-source ports, SQL catalogs, migrations | +| `@agentic-org/state` | generic state-store, outbox-source, inbox, idempotency, transaction, and lease ports | +| `@agentic-org/state-cockroach` | first replaceable durable SQL implementation of state-store and outbox-source ports | | `@agentic-org/messaging` | NATS envelope builder, subject builder, JetStream publisher, consumer, DLQ, replay contracts | | `@agentic-org/messaging-nats` | NATS JetStream implementation of the event publisher port, canonical JSON, headers, message IDs | | `@agentic-org/workflows-temporal` | Temporal workflow and activity contracts, task queues, workflow clients | @@ -241,9 +242,10 @@ commands should register a handler; new persistence backends should implement the same store-factory port. State-store ports are async at the application boundary. In-memory -adapters may resolve immediately, but CockroachDB, transactional -outbox, inbox, and lease adapters must be able to perform real I/O -without changing command-handler contracts. +adapters may resolve immediately, but durable SQL, transactional outbox, +inbox, and lease adapters must be able to perform real I/O without +changing command-handler contracts. CockroachDB is the first durable SQL +adapter in the cluster, not an application-layer dependency. ## SOLID Rules @@ -282,10 +284,11 @@ implement it. Runtime hosts bind implementations. All state transitions are event-producing commands. -CockroachDB stores authoritative state, audit, idempotency, and outbox. -NATS JetStream carries event distribution, inboxes, live UI updates, -replayable integration streams, and DLQs. Logs, traces, and metrics are -evidence. They are not business truth. +The durable state adapter stores authoritative state, audit, +idempotency, and outbox. In `full-ai-cluster`, the first implementation +is CockroachDB. NATS JetStream carries event distribution, inboxes, live +UI updates, replayable integration streams, and DLQs. Logs, traces, and +metrics are evidence. They are not business truth. ### Canonical Event Envelope @@ -589,7 +592,7 @@ tree. | Adapter package | Cluster dependency | Expected in-cluster target | | --------------------------------- | ------------------------------------------------ | --------------------------------------------------------------------- | -| `@agentic-org/state` | CockroachDB ArgoCD app | `cockroachdb-public.cockroachdb.svc.cluster.local:26257` | +| `@agentic-org/state-cockroach` | CockroachDB ArgoCD app | `cockroachdb-public.cockroachdb.svc.cluster.local:26257` | | `@agentic-org/messaging` | NATS ArgoCD app with JetStream enabled | `nats.nats.svc.cluster.local:4222` | | `@agentic-org/workflows-temporal` | Temporal ArgoCD app | `temporal-frontend.temporal.svc.cluster.local:7233` | | `@agentic-org/actors-dapr` | Dapr control plane | Dapr sidecar plus `dapr-system` placement service | @@ -678,7 +681,8 @@ They should get the narrowest network policy and credential scope first. Current cluster readiness: -- CockroachDB exists as the distributed SQL substrate. +- CockroachDB exists as the first distributed SQL substrate. It is + consumed only through the durable state adapter boundary. - NATS exists with JetStream enabled and Longhorn-backed file storage. - Temporal and Dapr are present, but their Organization-specific persistence/components still need wiring. @@ -700,11 +704,11 @@ each substrate becomes live. The same package architecture should run in three modes: -| Mode | Purpose | Runtime adapters | -| ----------------- | ------------------------------------------ | ------------------------------------------------------------------------------- | -| unit/test | package and command tests | in-memory/fake adapters | -| local dev cluster | k3d/K3S parity with `full-ai-cluster` apps | real NATS/Cockroach when available, fake Hermes/hat-system if needed | -| full cluster | production-like AI cluster | real CockroachDB, NATS, Hindsight, Hermes, OpenZiti, hat-system, Temporal, Dapr | +| Mode | Purpose | Runtime adapters | +| ----------------- | ------------------------------------------ | ----------------------------------------------------------------------------------------------- | +| unit/test | package and command tests | in-memory/fake adapters | +| local dev cluster | k3d/K3S parity with `full-ai-cluster` apps | real NATS/durable SQL when available, fake Hermes/hat-system if needed | +| full cluster | production-like AI cluster | CockroachDB-backed state adapter, NATS, Hindsight, Hermes, OpenZiti, hat-system, Temporal, Dapr | Do not create a Docker Compose architecture that diverges from `full-ai-cluster`. Local development can use fakes or a dev cluster, but @@ -781,8 +785,8 @@ other work. - `@agentic-org/ui-projections`. 2. Implement the canonical command context, event envelope, typed enums, and idempotency key builder. -3. Implement the first CockroachDB schema and Drizzle migrations for the - V0 executable contract. +3. Implement the first durable SQL schema and migrations for the V0 + executable contract, using CockroachDB as the initial adapter. 4. Implement command handlers for: - send supervisor signal; - triage supervisor signal; diff --git a/agentic-organization/docs/V0_EXECUTABLE_CONTRACT.md b/agentic-organization/docs/V0_EXECUTABLE_CONTRACT.md index 43fd25c498..957f6b87c3 100644 --- a/agentic-organization/docs/V0_EXECUTABLE_CONTRACT.md +++ b/agentic-organization/docs/V0_EXECUTABLE_CONTRACT.md @@ -18,20 +18,20 @@ not a parallel substrate. The current `origin/main` cluster shape gives V0 these host primitives: -| Cluster component | V0 use | -|---|---| -| K3S + ArgoCD App-of-Apps | deploy Agentic Organization as a future `full-ai-cluster/k8s/applications/agentic-organization/` application | -| Cilium + Hubble | pod networking, L7 policy, flow observability, and service-mesh behavior without Istio | -| cert-manager, Vault, SPIRE, Trust Manager, External Secrets | workload identity, TLS trust, and secret delivery | -| CockroachDB | authoritative Organization database | -| NATS JetStream | event transport, outbox fanout, live UI updates, and replayable integration streams | -| Temporal TS | durable workflows after the native command model is proven | -| Dapr Actors | hot entity coordination after the DB-backed service contract is proven | -| Hindsight | Hermes memory backend, wrapped with Organization attribution and scope | -| Hermes | agent runtime that performs the work | -| OZ/OpenZiti | zero-trust transport, not the Organization business orchestrator | -| hat-system | Kubernetes hat enforcement/projection surface using Hat, HatBinding, HatSwap, and HatPolicy CRDs | -| Loki, Tempo, Alloy, Mimir, kube-prometheus-stack | logs, traces, metrics, dashboards, and audit correlation | +| Cluster component | V0 use | +| ----------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | +| K3S + ArgoCD App-of-Apps | deploy Agentic Organization as a future `full-ai-cluster/k8s/applications/agentic-organization/` application | +| Cilium + Hubble | pod networking, L7 policy, flow observability, and service-mesh behavior without Istio | +| cert-manager, Vault, SPIRE, Trust Manager, External Secrets | workload identity, TLS trust, and secret delivery | +| CockroachDB | first durable SQL adapter for the authoritative Organization database boundary | +| NATS JetStream | event transport, outbox fanout, live UI updates, and replayable integration streams | +| Temporal TS | durable workflows after the native command model is proven | +| Dapr Actors | hot entity coordination after the DB-backed service contract is proven | +| Hindsight | Hermes memory backend, wrapped with Organization attribution and scope | +| Hermes | agent runtime that performs the work | +| OZ/OpenZiti | zero-trust transport, not the Organization business orchestrator | +| hat-system | Kubernetes hat enforcement/projection surface using Hat, HatBinding, HatSwap, and HatPolicy CRDs | +| Loki, Tempo, Alloy, Mimir, kube-prometheus-stack | logs, traces, metrics, dashboards, and audit correlation | Sync-wave implication: Agentic Organization is a consumer app. It should land after the foundation, data planes, hat-system CRDs, Hindsight, @@ -89,15 +89,15 @@ This is the smallest useful loop because it proves: Keep the first hat set small: -| Hat | V0 reason | -|---|---| -| Director | accepts or rejects escalated supervisor signals or capability requests for V0 scope | -| Engineering Manager | grooms the work item, selects schedule, assigns implementer and reviewer hats | -| Implementer | executes the prompt flow and submits evidence | -| Code Reviewer | reviews the evidence and blocks self-approval | -| Memory Curator | reviews memory writes or flags memory gaps when the run ends | -| Platform Operator | handles runtime failure, pod/session issues, and integration health | -| Security Reviewer | required only when the request needs a new credential or external tool scope | +| Hat | V0 reason | +| ------------------- | ----------------------------------------------------------------------------------- | +| Director | accepts or rejects escalated supervisor signals or capability requests for V0 scope | +| Engineering Manager | grooms the work item, selects schedule, assigns implementer and reviewer hats | +| Implementer | executes the prompt flow and submits evidence | +| Code Reviewer | reviews the evidence and blocks self-approval | +| Memory Curator | reviews memory writes or flags memory gaps when the run ends | +| Platform Operator | handles runtime failure, pod/session issues, and integration health | +| Security Reviewer | required only when the request needs a new credential or external tool scope | The Executive Board, TPM, Product Owner, Architect, QA Reviewer, Hat Designer, and department directors remain first-class in the reference @@ -196,7 +196,7 @@ the native service layer: ```text Temporal workflow or Dapr actor -> Organization command service - -> CockroachDB transaction + -> durable state transaction through the state adapter -> outbox event -> NATS publish -> trace, log, metric @@ -234,7 +234,8 @@ request -> ready gate -> hat assignment -> prompt-flow run The demo must show: -- CockroachDB state for the work item and assignment; +- durable state for the work item and assignment, backed by CockroachDB + in the first cluster adapter; - NATS/outbox events for every transition; - a discussion anchor tied to the work item; - a hat token with expiry; diff --git a/agentic-organization/docs/V0_SCHEMA_AND_COMMANDS.md b/agentic-organization/docs/V0_SCHEMA_AND_COMMANDS.md index ebc4de2c6c..e305c3148f 100644 --- a/agentic-organization/docs/V0_SCHEMA_AND_COMMANDS.md +++ b/agentic-organization/docs/V0_SCHEMA_AND_COMMANDS.md @@ -7,27 +7,30 @@ contract for Agentic Organization. It is not a full DDL. It is the shape the domain model, Drizzle migrations, command handlers, MCP tools, workers, and tests should agree on before implementation starts. -CockroachDB is the authoritative store for Organization-owned state. +The durable state adapter is the authoritative store for +Organization-owned state. CockroachDB is the first implementation +because it exists in `full-ai-cluster`, but application code must depend +on generic state ports so another database can replace it later. Temporal history, Dapr actor state, NATS streams, Hindsight memory, and hat-system CRDs are runtime surfaces or projections. They do not replace -the Organization database. +the Organization database boundary. ## Global Columns Every authoritative table should include: -| Column | Purpose | -|---|---| -| `id` | stable unique ID | -| `organization_id` | future multi-org partition key, even if V0 uses one org | -| `created_at` | creation time | -| `updated_at` | last mutation time | -| `version` | optimistic concurrency and projection safety | -| `created_by_agent_id` | agent that caused the write, when applicable | -| `created_by_hat_assignment_id` | hat authority that caused the write, when applicable | -| `correlation_id` | end-to-end request/run correlation | -| `causation_id` | direct parent command, event, tool call, or workflow step | -| `trace_id` | observability trace link | +| Column | Purpose | +| ------------------------------ | --------------------------------------------------------- | +| `id` | stable unique ID | +| `organization_id` | future multi-org partition key, even if V0 uses one org | +| `created_at` | creation time | +| `updated_at` | last mutation time | +| `version` | optimistic concurrency and projection safety | +| `created_by_agent_id` | agent that caused the write, when applicable | +| `created_by_hat_assignment_id` | hat authority that caused the write, when applicable | +| `correlation_id` | end-to-end request/run correlation | +| `causation_id` | direct parent command, event, tool call, or workflow step | +| `trace_id` | observability trace link | Append-only records should also carry `sequence` when replay order matters. @@ -36,81 +39,81 @@ matters. ### Identity and Hats -| Table | V0 responsibility | -|---|---| -| `agents` | known Hermes-capable agents and their stable identity | -| `agent_sessions` | live or historical Hermes sessions bound to an agent | -| `departments` | first department containers for ownership and review routing | -| `hat_definitions` | Organization-owned hat catalog | -| `hat_authority_rules` | typed permissions, scopes, and policy metadata for a hat | -| `hat_skill_bindings` | skills and prompt-flow availability attached to a hat | -| `hat_supply_policies` | max concurrency, TTL, cooldown, warmup, and assignment rules | -| `hat_assignments` | time-bounded wearer assignment for a specific agent/session | -| `hat_tokens` | short-lived JWT issuance, refresh, revocation, and expiry state | +| Table | V0 responsibility | +| ------------------------ | --------------------------------------------------------------------------- | +| `agents` | known Hermes-capable agents and their stable identity | +| `agent_sessions` | live or historical Hermes sessions bound to an agent | +| `departments` | first department containers for ownership and review routing | +| `hat_definitions` | Organization-owned hat catalog | +| `hat_authority_rules` | typed permissions, scopes, and policy metadata for a hat | +| `hat_skill_bindings` | skills and prompt-flow availability attached to a hat | +| `hat_supply_policies` | max concurrency, TTL, cooldown, warmup, and assignment rules | +| `hat_assignments` | time-bounded wearer assignment for a specific agent/session | +| `hat_tokens` | short-lived JWT issuance, refresh, revocation, and expiry state | | `hat_system_projections` | last observed Hat, HatBinding, HatSwap, and HatPolicy state from Kubernetes | ### Work Management -| Table | V0 responsibility | -|---|---| -| `projects` | top-level work containers | -| `initiatives` | project-scoped bodies of work | -| `work_items` | supervisor-chain signals, capability requests, tasks, defects, reviews, and follow-ups | -| `work_item_state_history` | append-only state transitions | -| `work_item_dependencies` | blocking or informational dependencies | -| `blockers` | active impediments with owner, severity, and resolution path | -| `assignments` | work item to agent/hat/session assignment records | -| `gates` | required review points for readiness, code, QA, memory, or security | -| `gate_decisions` | approve, reject, needs-changes, or defer decisions | -| `releases` | release groupings once release management enters the slice | +| Table | V0 responsibility | +| ------------------------- | -------------------------------------------------------------------------------------- | +| `projects` | top-level work containers | +| `initiatives` | project-scoped bodies of work | +| `work_items` | supervisor-chain signals, capability requests, tasks, defects, reviews, and follow-ups | +| `work_item_state_history` | append-only state transitions | +| `work_item_dependencies` | blocking or informational dependencies | +| `blockers` | active impediments with owner, severity, and resolution path | +| `assignments` | work item to agent/hat/session assignment records | +| `gates` | required review points for readiness, code, QA, memory, or security | +| `gate_decisions` | approve, reject, needs-changes, or defer decisions | +| `releases` | release groupings once release management enters the slice | ### Schedules, Prompt Flows, and Actions -| Table | V0 responsibility | -|---|---| -| `hat_schedule_templates` | default work rhythm by hat | -| `work_schedules` | concrete schedule assigned to an agent/hat context | -| `work_schedule_blocks` | free time, prioritized work, review, reflection, or meeting blocks | -| `prompt_flow_definitions` | named deterministic work protocols | -| `prompt_flow_versions` | immutable versioned prompt-flow contract | -| `prompt_flow_phases` | ordered reusable phases | -| `hat_prompt_flow_bindings` | which hats can run which prompt flows | -| `prompt_flow_runs` | one execution of a prompt-flow version | -| `prompt_flow_phase_runs` | state and evidence for each phase execution | -| `prompt_flow_gate_decisions` | reviewer decisions at phase boundaries | -| `universal_action_definitions` | typed action grammar catalog | -| `universal_action_records` | action intent emitted by an agent or workflow | -| `universal_action_observations` | observed result, evidence, and side effects for an action | +| Table | V0 responsibility | +| ------------------------------- | ------------------------------------------------------------------ | +| `hat_schedule_templates` | default work rhythm by hat | +| `work_schedules` | concrete schedule assigned to an agent/hat context | +| `work_schedule_blocks` | free time, prioritized work, review, reflection, or meeting blocks | +| `prompt_flow_definitions` | named deterministic work protocols | +| `prompt_flow_versions` | immutable versioned prompt-flow contract | +| `prompt_flow_phases` | ordered reusable phases | +| `hat_prompt_flow_bindings` | which hats can run which prompt flows | +| `prompt_flow_runs` | one execution of a prompt-flow version | +| `prompt_flow_phase_runs` | state and evidence for each phase execution | +| `prompt_flow_gate_decisions` | reviewer decisions at phase boundaries | +| `universal_action_definitions` | typed action grammar catalog | +| `universal_action_records` | action intent emitted by an agent or workflow | +| `universal_action_observations` | observed result, evidence, and side effects for an action | ### Communication, Graph, Documents, and Context -| Table | V0 responsibility | -|---|---| -| `supervisor_signals` | supervisor-chain and capability-request intake records before or during work-item routing | -| `discussion_anchors` | required work/project/initiative/task anchor for any discussion | -| `conversation_threads` | one-on-one, team, department, executive, or broadcast thread | -| `messages` | immutable message log with actor and hat attribution | -| `meetings` | structured meeting sessions with mode and anchor | -| `decisions` | explicit decisions linked to work and evidence | -| `documents` | BRDs, CAs, ADRs, reports, test cases, runbooks, and memory reviews | -| `artifact_links` | logs, screenshots, traces, code refs, PRs, builds, and uploads | -| `graph_nodes` | agent-readable graph node registry | -| `graph_edges` | typed relationships between work, docs, messages, decisions, runs, and memories | -| `context_packs` | deterministic context bundles assembled for an agent run or review | +| Table | V0 responsibility | +| ---------------------- | ----------------------------------------------------------------------------------------- | +| `supervisor_signals` | supervisor-chain and capability-request intake records before or during work-item routing | +| `discussion_anchors` | required work/project/initiative/task anchor for any discussion | +| `conversation_threads` | one-on-one, team, department, executive, or broadcast thread | +| `messages` | immutable message log with actor and hat attribution | +| `meetings` | structured meeting sessions with mode and anchor | +| `decisions` | explicit decisions linked to work and evidence | +| `documents` | BRDs, CAs, ADRs, reports, test cases, runbooks, and memory reviews | +| `artifact_links` | logs, screenshots, traces, code refs, PRs, builds, and uploads | +| `graph_nodes` | agent-readable graph node registry | +| `graph_edges` | typed relationships between work, docs, messages, decisions, runs, and memories | +| `context_packs` | deterministic context bundles assembled for an agent run or review | ### Runtime, Memory, Security, and Audit -| Table | V0 responsibility | -|---|---| -| `hermes_runs` | Organization binding to a Hermes execution session | -| `mcp_tool_calls` | governed tool call attempts and results | -| `memory_events` | Hindsight recall, retain, reflect, and review attribution | -| `credential_requests` | requests to expand credential proxy or external tool scope | -| `signals` | durable internal signals consumed by workers and UI read models | -| `audit_events` | append-only policy and state-change audit trail | -| `outbox_events` | transactional event publication source for NATS | -| `runtime_leases` | scheduler, reconciler, and worker leases | -| `idempotency_keys` | command deduplication records | +| Table | V0 responsibility | +| --------------------- | --------------------------------------------------------------- | +| `hermes_runs` | Organization binding to a Hermes execution session | +| `mcp_tool_calls` | governed tool call attempts and results | +| `memory_events` | Hindsight recall, retain, reflect, and review attribution | +| `credential_requests` | requests to expand credential proxy or external tool scope | +| `signals` | durable internal signals consumed by workers and UI read models | +| `audit_events` | append-only policy and state-change audit trail | +| `outbox_events` | transactional event publication source for NATS | +| `runtime_leases` | scheduler, reconciler, and worker leases | +| `idempotency_keys` | command deduplication records | ## V0 Enums @@ -258,7 +261,7 @@ Every side-effecting command must include: Every command handler must: -1. load authoritative state from CockroachDB; +1. load authoritative state through the state-store port; 2. validate actor context and hat authority; 3. validate lifecycle transition; 4. write state, audit event, and outbox event in one transaction; @@ -267,29 +270,29 @@ Every command handler must: ## V0 Commands -| Command | Actor scope | Writes | Emits | -|---|---|---|---| -| `send_supervisor_signal` | any authorized hat with supervisor line; capability request inputs enter through this path | `supervisor_signals`, `work_items`, `discussion_anchors`, `graph_nodes`, `audit_events`, `outbox_events` | `supervisor_signal_sent`, `work_item_changed` | -| `triage_supervisor_signal` | target supervisor hat, director, or engineering manager | `supervisor_signals`, `work_items`, `assignments`, `gates`, `context_packs` | `supervisor_signal_triaged`, `work_item_changed`, `gate_requested` | -| `create_discussion_anchor` | any authorized hat | `discussion_anchors`, `graph_edges` | `work_item_changed` | -| `create_context_pack` | manager, reviewer, implementer for assigned work | `context_packs`, `graph_edges`, `audit_events` | `work_item_changed` | -| `mark_work_ready` | manager or reviewer | `work_items`, `work_item_state_history`, `gates` | `work_item_changed`, `gate_requested` | -| `reserve_hat` | manager, director, platform operator | `hat_assignments`, `hat_tokens`, `audit_events` | `hat_assignment_changed` | -| `issue_hat_token` | hat service, after policy allow | `hat_tokens`, `audit_events` | `hat_token_changed` | -| `refresh_hat_token` | active assigned agent/session | `hat_tokens`, `audit_events` | `hat_token_changed` | -| `revoke_hat_assignment` | manager, director, security, policy automation | `hat_assignments`, `hat_tokens`, `audit_events` | `hat_assignment_changed`, `hat_token_changed` | -| `start_schedule_block` | assigned agent/session or scheduler | `work_schedule_blocks`, `agent_sessions` | `schedule_block_changed` | -| `start_prompt_flow` | assigned agent/session | `prompt_flow_runs`, `prompt_flow_phase_runs` | `prompt_flow_changed` | -| `record_universal_action` | assigned agent/session, workflow activity, adapter | `universal_action_records`, `mcp_tool_calls`, `audit_events` | `prompt_flow_changed` | -| `record_action_observation` | adapter, worker, reviewer, assigned agent | `universal_action_observations`, `artifact_links` | `prompt_flow_changed` | -| `launch_hermes_run` | runtime service or Temporal activity | `hermes_runs`, `agent_sessions`, `audit_events` | `hermes_run_changed` | -| `record_hermes_run_status` | Hermes/OZ callback, reconciler, platform operator | `hermes_runs`, `artifact_links` | `hermes_run_changed` | -| `submit_evidence` | implementer, QA, reviewer, adapter | `artifact_links`, `graph_edges`, `audit_events` | `work_item_changed` | -| `request_gate_review` | implementer, manager, workflow | `gates`, `work_items` | `gate_requested`, `work_item_changed` | -| `decide_gate` | reviewer hat, not same active implementer assignment | `gate_decisions`, `gates`, `work_items`, `audit_events` | `gate_decided`, `work_item_changed` | -| `record_memory_event` | memory adapter, assigned agent/session, memory curator | `memory_events`, `graph_edges`, `audit_events` | `memory_event_recorded` | -| `submit_credential_request` | any authorized hat with anchored work | `credential_requests`, `work_items`, `discussion_anchors` | `credential_request_changed` | -| `complete_outcome_review` | manager, memory curator, reviewer | `work_items`, `decisions`, optional follow-up `work_items` | `outcome_review_completed` | +| Command | Actor scope | Writes | Emits | +| --------------------------- | ------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------ | +| `send_supervisor_signal` | any authorized hat with supervisor line; capability request inputs enter through this path | `supervisor_signals`, `work_items`, `discussion_anchors`, `graph_nodes`, `audit_events`, `outbox_events` | `supervisor_signal_sent`, `work_item_changed` | +| `triage_supervisor_signal` | target supervisor hat, director, or engineering manager | `supervisor_signals`, `work_items`, `assignments`, `gates`, `context_packs` | `supervisor_signal_triaged`, `work_item_changed`, `gate_requested` | +| `create_discussion_anchor` | any authorized hat | `discussion_anchors`, `graph_edges` | `work_item_changed` | +| `create_context_pack` | manager, reviewer, implementer for assigned work | `context_packs`, `graph_edges`, `audit_events` | `work_item_changed` | +| `mark_work_ready` | manager or reviewer | `work_items`, `work_item_state_history`, `gates` | `work_item_changed`, `gate_requested` | +| `reserve_hat` | manager, director, platform operator | `hat_assignments`, `hat_tokens`, `audit_events` | `hat_assignment_changed` | +| `issue_hat_token` | hat service, after policy allow | `hat_tokens`, `audit_events` | `hat_token_changed` | +| `refresh_hat_token` | active assigned agent/session | `hat_tokens`, `audit_events` | `hat_token_changed` | +| `revoke_hat_assignment` | manager, director, security, policy automation | `hat_assignments`, `hat_tokens`, `audit_events` | `hat_assignment_changed`, `hat_token_changed` | +| `start_schedule_block` | assigned agent/session or scheduler | `work_schedule_blocks`, `agent_sessions` | `schedule_block_changed` | +| `start_prompt_flow` | assigned agent/session | `prompt_flow_runs`, `prompt_flow_phase_runs` | `prompt_flow_changed` | +| `record_universal_action` | assigned agent/session, workflow activity, adapter | `universal_action_records`, `mcp_tool_calls`, `audit_events` | `prompt_flow_changed` | +| `record_action_observation` | adapter, worker, reviewer, assigned agent | `universal_action_observations`, `artifact_links` | `prompt_flow_changed` | +| `launch_hermes_run` | runtime service or Temporal activity | `hermes_runs`, `agent_sessions`, `audit_events` | `hermes_run_changed` | +| `record_hermes_run_status` | Hermes/OZ callback, reconciler, platform operator | `hermes_runs`, `artifact_links` | `hermes_run_changed` | +| `submit_evidence` | implementer, QA, reviewer, adapter | `artifact_links`, `graph_edges`, `audit_events` | `work_item_changed` | +| `request_gate_review` | implementer, manager, workflow | `gates`, `work_items` | `gate_requested`, `work_item_changed` | +| `decide_gate` | reviewer hat, not same active implementer assignment | `gate_decisions`, `gates`, `work_items`, `audit_events` | `gate_decided`, `work_item_changed` | +| `record_memory_event` | memory adapter, assigned agent/session, memory curator | `memory_events`, `graph_edges`, `audit_events` | `memory_event_recorded` | +| `submit_credential_request` | any authorized hat with anchored work | `credential_requests`, `work_items`, `discussion_anchors` | `credential_request_changed` | +| `complete_outcome_review` | manager, memory curator, reviewer | `work_items`, `decisions`, optional follow-up `work_items` | `outcome_review_completed` | ## Idempotency @@ -321,9 +324,10 @@ idempotency conflict. ## Outbox and NATS -CockroachDB transactions should write domain state and `outbox_events` -together. A worker publishes outbox rows to NATS JetStream and marks -them published. +Durable state transactions should write domain state and `outbox_events` +together. The first durable adapter uses CockroachDB, but the command +model only depends on generic state ports. A worker publishes outbox +rows to NATS JetStream and marks them published. Subject shape: diff --git a/agentic-organization/packages/README.md b/agentic-organization/packages/README.md index fd0bbea7dc..b3d8a6ed20 100644 --- a/agentic-organization/packages/README.md +++ b/agentic-organization/packages/README.md @@ -11,8 +11,8 @@ host, or Kubernetes deployment is introduced. | ----------------- | -------------------------------------------------------------------------------------------------------------------------- | | `domain` | typed command names, event names, aggregate names, work item state machine, event envelope, shared records | | `application` | command pipeline, handler registry, idempotency handling, state-store ports, first supervisor-chain signal command handler | -| `state` | in-memory Organization state-store factory used as the first repository port fake | -| `state-cockroach` | CockroachDB state-store/outbox-source contracts, SQL statement catalogs, and first core-state migration skeleton | +| `state` | generic state-store and outbox-source ports plus the in-memory Organization state-store factory fake | +| `state-cockroach` | first replaceable durable SQL adapter for the state-store/outbox-source ports, backed by CockroachDB | | `messaging` | NATS subject contract, outbox publisher port, event publisher port, and domain resolver without a live NATS dependency | | `messaging-nats` | NATS JetStream event publisher adapter contract with canonical JSON payloads, headers, and message IDs | | `observability` | LGTM/OpenTelemetry attribute projection from Agentic event envelopes | @@ -38,7 +38,10 @@ supervisor-chain signal command CockroachDB, JetStream publishing, Temporal, Dapr, Hermes, Hindsight, and the hat-system CRDs come next as adapters behind these contracts. They should not redefine command names, event names, state names, -correlation fields, or policy authority. +correlation fields, or policy authority. CockroachDB is the first +durable state adapter because it exists in `full-ai-cluster`; application +and messaging code must remain database-agnostic so a later durable +store can replace it behind the same ports. The application package must not construct concrete state adapters. Runtime hosts and tests provide a `CommandStateStoreFactory`; the state @@ -46,16 +49,17 @@ package implements the current in-memory factory. Command routing uses a handler registry so new commands add handlers instead of editing a central `switch` or `if` dispatcher. -`CommandStateStore` is async even when backed by the in-memory fake. The -real CockroachDB adapter must not be squeezed into a synchronous toy -shape. +`CommandStateStore` and `OutboxEventSource` are async even when backed +by in-memory fakes. Durable adapters must not be squeezed into a +synchronous toy shape. The outbox publisher owns the generic publish loop: claim unpublished -outbox rows, resolve a typed Organization domain, publish through an -`EventPublisher` port, and mark the outbox row published only after the -publish returns successfully. The NATS package implements that port and -is the only package in this slice that knows about NATS headers, -message IDs, and JSON transport payloads. +outbox rows from the generic state port, resolve a typed Organization +domain, publish through an `EventPublisher` port, and mark the outbox +row published only after the publish returns successfully. The NATS +package implements that publisher port and is the only package in this +slice that knows about NATS headers, message IDs, and JSON transport +payloads. State adapters must not import messaging adapters. ## Validation diff --git a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts index b9ae9bdd1e..499b70d246 100644 --- a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts +++ b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts @@ -30,6 +30,11 @@ describe("package dependency boundaries", () => { sourceGlob: "messaging/src/**/*.ts", forbiddenImportFragments: ["../messaging-nats", "../../messaging-nats", "nats"], }, + { + packageName: PackageBoundaryRule.StateAdapter, + sourceGlob: "state-cockroach/src/**/*.ts", + forbiddenImportFragments: ["../../messaging", "../messaging", "nats", "jetstream"], + }, ], }); diff --git a/agentic-organization/packages/governance/src/package-dependency-boundaries.ts b/agentic-organization/packages/governance/src/package-dependency-boundaries.ts index 08bfb7386a..eb7a82d16c 100644 --- a/agentic-organization/packages/governance/src/package-dependency-boundaries.ts +++ b/agentic-organization/packages/governance/src/package-dependency-boundaries.ts @@ -5,6 +5,7 @@ import { fileURLToPath } from "node:url"; export const PackageBoundaryRule = { Application: "application", Messaging: "messaging", + StateAdapter: "state_adapter", } as const; export type PackageBoundaryRule = (typeof PackageBoundaryRule)[keyof typeof PackageBoundaryRule]; diff --git a/agentic-organization/packages/messaging/src/index.ts b/agentic-organization/packages/messaging/src/index.ts index 2cb19ab88e..87b9ed3716 100644 --- a/agentic-organization/packages/messaging/src/index.ts +++ b/agentic-organization/packages/messaging/src/index.ts @@ -4,12 +4,9 @@ export { OutboxPublishOutcomeStatus, createOutboxPublisher, resolveAgenticMessagingDomain, - type ClaimUnpublishedOutboxEventsInput, type CreateOutboxPublisherInput, type EventPublication, type EventPublisher, - type MarkOutboxEventPublishedInput, - type OutboxEventSource, type OutboxPublishBatchResult, type OutboxPublisher, type ResolveAgenticMessagingDomain, diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.test.ts b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts index 4de151fdc3..d47415bd54 100644 --- a/agentic-organization/packages/messaging/src/outbox-publisher.test.ts +++ b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts @@ -7,6 +7,7 @@ import { createAgenticEventEnvelope, type OutboxEvent, } from "../../domain/src/index.ts"; +import type { OutboxEventSource } from "../../state/src/index.ts"; import { AgenticMessagingDomain, OutboxPublishOutcomeStatus, @@ -14,7 +15,6 @@ import { resolveAgenticMessagingDomain, type EventPublication, type EventPublisher, - type OutboxEventSource, } from "./outbox-publisher.ts"; describe("outbox publisher", () => { diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.ts b/agentic-organization/packages/messaging/src/outbox-publisher.ts index 0897a35274..50209ff9d3 100644 --- a/agentic-organization/packages/messaging/src/outbox-publisher.ts +++ b/agentic-organization/packages/messaging/src/outbox-publisher.ts @@ -1,4 +1,5 @@ import { AgenticEventType, type OutboxEvent } from "../../domain/src/index.ts"; +import type { ClaimUnpublishedOutboxEventsInput, OutboxEventSource } from "../../state/src/index.ts"; import { buildAgenticEventSubject } from "./subject-builder.ts"; export const AgenticMessagingDomain = { @@ -15,20 +16,6 @@ export const OutboxPublishOutcomeStatus = { export type OutboxPublishOutcomeStatus = (typeof OutboxPublishOutcomeStatus)[keyof typeof OutboxPublishOutcomeStatus]; -export type ClaimUnpublishedOutboxEventsInput = { - batchSize: number; -}; - -export type MarkOutboxEventPublishedInput = { - outboxEventId: string; - publishedAt: string; -}; - -export type OutboxEventSource = { - claimUnpublishedOutboxEvents: (input: ClaimUnpublishedOutboxEventsInput) => Promise; - markOutboxEventPublished: (input: MarkOutboxEventPublishedInput) => Promise; -}; - export type EventPublication = { subject: string; outboxEvent: OutboxEvent; diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts index 28b74733ba..9234b21277 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts @@ -1,5 +1,5 @@ import type { AgenticEventEnvelope } from "../../domain/src/index.ts"; -import type { OutboxEventSource } from "../../messaging/src/index.ts"; +import type { OutboxEventSource } from "../../state/src/index.ts"; import { CockroachTableName } from "./cockroach-schema.ts"; export const CockroachOutboxEventSourceStatement = { diff --git a/agentic-organization/packages/state/src/index.ts b/agentic-organization/packages/state/src/index.ts index 8ab8d1eff9..1b1611aa83 100644 --- a/agentic-organization/packages/state/src/index.ts +++ b/agentic-organization/packages/state/src/index.ts @@ -3,3 +3,8 @@ export { type InMemoryOrganizationStoreFactory, type InMemoryOrganizationStoreSnapshot, } from "./in-memory-organization-store.ts"; +export type { + ClaimUnpublishedOutboxEventsInput, + MarkOutboxEventPublishedInput, + OutboxEventSource, +} from "./outbox-event-source.ts"; diff --git a/agentic-organization/packages/state/src/outbox-event-source.ts b/agentic-organization/packages/state/src/outbox-event-source.ts new file mode 100644 index 0000000000..6125bed5f4 --- /dev/null +++ b/agentic-organization/packages/state/src/outbox-event-source.ts @@ -0,0 +1,15 @@ +import type { OutboxEvent } from "../../domain/src/index.ts"; + +export type ClaimUnpublishedOutboxEventsInput = { + batchSize: number; +}; + +export type MarkOutboxEventPublishedInput = { + outboxEventId: string; + publishedAt: string; +}; + +export type OutboxEventSource = { + claimUnpublishedOutboxEvents: (input: ClaimUnpublishedOutboxEventsInput) => Promise; + markOutboxEventPublished: (input: MarkOutboxEventPublishedInput) => Promise; +}; diff --git a/openspec/specs/agentic-organization/spec.md b/openspec/specs/agentic-organization/spec.md index 71f30853f1..a3d752c6bc 100644 --- a/openspec/specs/agentic-organization/spec.md +++ b/openspec/specs/agentic-organization/spec.md @@ -56,21 +56,32 @@ Organization state only by calling Organization commands. concrete state adapters, Cockroach adapters, NestJS, NATS, Dapr, Temporal, Drizzle, Postgres, or other runtime clients - **AND** a violation fails the test suite before the boundary can drift +- **AND** state adapter source files are checked for forbidden imports + of messaging, NATS, JetStream, or other event transport clients ### Requirement: Commands are idempotent Organization commands MUST use deterministic idempotency keys at the command boundary. -#### Scenario: Cockroach core state schema exists +#### Scenario: Durable core state schema exists -- **WHEN** the first CockroachDB migration contract is loaded +- **WHEN** the first durable state migration contract is loaded - **THEN** it declares work item, supervisor signal, audit event, outbox event, and idempotency record tables - **AND** outbox rows include trace ID, correlation ID, and canonical envelope JSON fields for later NATS publication and workflow visibility +#### Scenario: Durable state adapter is replaceable + +- **WHEN** application or messaging package source is inspected +- **THEN** it does not import CockroachDB, Drizzle, Postgres, or + database-client packages +- **AND** it depends on generic state and outbox-source ports instead +- **AND** CockroachDB is treated as the first replaceable durable adapter + for the cluster, not as the application model + #### Scenario: Matching replay - **WHEN** a command is submitted twice with the same idempotency key From 50e6dd256b201bc26ee98d271a02e0892c95d9ef Mon Sep 17 00:00:00 2001 From: Lior Date: Mon, 25 May 2026 18:13:33 -0400 Subject: [PATCH 4/5] release: task-pr-4990-nats-outbox-replay - opened PR #5007 Co-Authored-By: Codex --- .../claims/task-pr-4990-nats-outbox-replay.md | 20 ------------------- 1 file changed, 20 deletions(-) delete mode 100644 docs/claims/task-pr-4990-nats-outbox-replay.md diff --git a/docs/claims/task-pr-4990-nats-outbox-replay.md b/docs/claims/task-pr-4990-nats-outbox-replay.md deleted file mode 100644 index f7788529a0..0000000000 --- a/docs/claims/task-pr-4990-nats-outbox-replay.md +++ /dev/null @@ -1,20 +0,0 @@ -# Claim - task-pr-4990-nats-outbox-replay - -- **Session ID:** codex/20260525T2209Z -- **Harness:** codex -- **Claimed at:** 2026-05-25T22:09:40Z -- **ETA:** 2026-05-25T22:30:00Z -- **Scope:** Replay only the intended post-#4990 agentic-organization NATS outbox and replaceable state-adapter slice onto current main, excluding stale-base governance/backlog deletions. -- **Durable target:** draft PR from a clean Codex branch after replay and focused package checks -- **Platform mirror:** https://github.com/Lucent-Financial-Group/Zeta/pull/4990 - -## Notes - -- Source branch under triage: `codex/agentic-org-package-ca-clean` at - `16417aecba9eae921bba0e363c2f5f9c4d21bbe7` as of - 2026-05-25T22:09Z. -- Do not delete or force-update that source branch during this claim. -- Current source branch diff includes stale-base noise outside scope: - `GOVERNANCE.md`, `docs/BACKLOG.md`, and a backlog row deletion. -- Dirty older #4990 local worktrees are out of scope and must remain untouched - unless a later tick explicitly proves they are safe to clean. From 779720fb5934d3724d9bf3ac2fdf6cadfbfcc2e4 Mon Sep 17 00:00:00 2001 From: Lior Date: Mon, 25 May 2026 18:30:51 -0400 Subject: [PATCH 5/5] fix(agentic-org): address outbox publisher review Make outbox claiming atomic, keep publish marking idempotent, align messaging subjects with the broader work domain, and extend messaging boundary coverage for DB client imports. Co-Authored-By: Codex --- .../src/package-dependency-boundaries.test.ts | 9 ++++- .../nats-jetstream-event-publisher.test.ts | 4 +-- .../messaging/src/outbox-publisher.test.ts | 4 +-- .../messaging/src/outbox-publisher.ts | 7 ++-- .../0001_agentic_org_core_state.sql | 2 ++ .../src/cockroach-outbox-event-source.test.ts | 36 ++++++++++++++++--- .../src/cockroach-outbox-event-source.ts | 35 ++++++++++++++---- .../src/cockroach-schema.test.ts | 2 ++ .../state-cockroach/src/cockroach-schema.ts | 2 ++ 9 files changed, 80 insertions(+), 21 deletions(-) diff --git a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts index 499b70d246..10a3f3c892 100644 --- a/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts +++ b/agentic-organization/packages/governance/src/package-dependency-boundaries.test.ts @@ -28,7 +28,14 @@ describe("package dependency boundaries", () => { { packageName: PackageBoundaryRule.Messaging, sourceGlob: "messaging/src/**/*.ts", - forbiddenImportFragments: ["../messaging-nats", "../../messaging-nats", "nats"], + forbiddenImportFragments: [ + "../messaging-nats", + "../../messaging-nats", + "nats", + "drizzle", + "pg", + "postgres", + ], }, { packageName: PackageBoundaryRule.StateAdapter, diff --git a/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts index b565bdc65f..8a56b42b06 100644 --- a/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts +++ b/agentic-organization/packages/messaging-nats/src/nats-jetstream-event-publisher.test.ts @@ -22,13 +22,13 @@ describe("NATS JetStream event publisher", () => { const outboxEvent = createOutboxEvent(); await publisher.publish({ - subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + subject: "agentic-org.local.org-lfg.work.supervisor_signal.sent", outboxEvent, }); equal(client.messages.length, 1); deepEqual(client.messages[0], { - subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + subject: "agentic-org.local.org-lfg.work.supervisor_signal.sent", payload: JSON.stringify(outboxEvent.envelope), messageId: "evt-001", headers: { diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.test.ts b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts index d47415bd54..23199bb9f8 100644 --- a/agentic-organization/packages/messaging/src/outbox-publisher.test.ts +++ b/agentic-organization/packages/messaging/src/outbox-publisher.test.ts @@ -21,7 +21,7 @@ describe("outbox publisher", () => { test("resolves event domains through typed mappings", () => { deepEqual( resolveAgenticMessagingDomain(AgenticEventType.SupervisorSignalSent), - AgenticMessagingDomain.SupervisorSignal, + AgenticMessagingDomain.Work, ); }); @@ -54,7 +54,7 @@ describe("outbox publisher", () => { ]); deepEqual(eventPublisher.publications, [ { - subject: "agentic-org.local.org-lfg.supervisor_signal.supervisor_signal.sent", + subject: "agentic-org.local.org-lfg.work.supervisor_signal.sent", outboxEvent, }, ]); diff --git a/agentic-organization/packages/messaging/src/outbox-publisher.ts b/agentic-organization/packages/messaging/src/outbox-publisher.ts index 50209ff9d3..03422f226f 100644 --- a/agentic-organization/packages/messaging/src/outbox-publisher.ts +++ b/agentic-organization/packages/messaging/src/outbox-publisher.ts @@ -3,8 +3,7 @@ import type { ClaimUnpublishedOutboxEventsInput, OutboxEventSource } from "../.. import { buildAgenticEventSubject } from "./subject-builder.ts"; export const AgenticMessagingDomain = { - SupervisorSignal: "supervisor_signal", - WorkItem: "work_item", + Work: "work", } as const; export type AgenticMessagingDomain = (typeof AgenticMessagingDomain)[keyof typeof AgenticMessagingDomain]; @@ -90,11 +89,11 @@ export function createOutboxPublisher(input: CreateOutboxPublisherInput): Outbox export function resolveAgenticMessagingDomain(eventType: AgenticEventType): AgenticMessagingDomain { if (eventType === AgenticEventType.SupervisorSignalSent) { - return AgenticMessagingDomain.SupervisorSignal; + return AgenticMessagingDomain.Work; } if (eventType === AgenticEventType.WorkItemChanged || eventType === AgenticEventType.WorkItemStateChanged) { - return AgenticMessagingDomain.WorkItem; + return AgenticMessagingDomain.Work; } throw new Error(`unsupported event type for messaging domain: ${eventType}`); diff --git a/agentic-organization/packages/state-cockroach/migrations/0001_agentic_org_core_state.sql b/agentic-organization/packages/state-cockroach/migrations/0001_agentic_org_core_state.sql index d905f5cf3c..6cbc38010b 100644 --- a/agentic-organization/packages/state-cockroach/migrations/0001_agentic_org_core_state.sql +++ b/agentic-organization/packages/state-cockroach/migrations/0001_agentic_org_core_state.sql @@ -47,6 +47,8 @@ CREATE TABLE IF NOT EXISTS agentic_org_outbox_events ( trace_id STRING NOT NULL, correlation_id STRING NOT NULL, envelope_json JSONB NOT NULL, + claimed_at TIMESTAMPTZ, + claim_expires_at TIMESTAMPTZ, published_at TIMESTAMPTZ ); diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts index a5e6eda061..b332ad205f 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.test.ts @@ -1,4 +1,4 @@ -import { deepEqual } from "node:assert/strict"; +import { deepEqual, equal, ok } from "node:assert/strict"; import { describe, test } from "node:test"; import { AgenticAggregateType, AgenticEventType, type AgenticEventEnvelope } from "../../domain/src/index.ts"; @@ -37,15 +37,41 @@ describe("cockroach outbox event source", () => { CockroachOutboxEventSourceStatement.MarkOutboxEventPublished, ], ); + ok(/UPDATE agentic_org_outbox_events/.test(executor.statements[0]?.sql ?? "")); + ok(/FOR UPDATE SKIP LOCKED/.test(executor.statements[0]?.sql ?? "")); + ok(/AND published_at IS NULL/.test(executor.statements[1]?.sql ?? "")); + }); + + test("rejects stale or duplicate publish marks", async () => { + const executor = createRecordingExecutor({ + markRows: [], + }); + const outboxSource = createCockroachOutboxEventSource({ + executor, + }); + + try { + await outboxSource.markOutboxEventPublished({ + outboxEventId: "outbox-001", + publishedAt: "2026-05-25T21:00:00.000Z", + }); + throw new Error("expected duplicate publish mark to reject"); + } catch (error) { + equal((error as Error).message, "outbox event was already published or missing: outbox-001"); + } }); }); type RecordingCockroachOutboxSqlExecutor = CockroachOutboxSqlExecutor & { - statements: { name: CockroachOutboxEventSourceStatement; parameters: readonly unknown[] }[]; + statements: { name: CockroachOutboxEventSourceStatement; sql: string; parameters: readonly unknown[] }[]; +}; + +type CreateRecordingExecutorInput = { + markRows?: { outbox_event_id: string }[]; }; -function createRecordingExecutor(): RecordingCockroachOutboxSqlExecutor { - const statements: { name: CockroachOutboxEventSourceStatement; parameters: readonly unknown[] }[] = []; +function createRecordingExecutor(input: CreateRecordingExecutorInput = {}): RecordingCockroachOutboxSqlExecutor { + const statements: { name: CockroachOutboxEventSourceStatement; sql: string; parameters: readonly unknown[] }[] = []; return { statements, @@ -64,7 +90,7 @@ function createRecordingExecutor(): RecordingCockroachOutboxSqlExecutor { } return { - rows: [], + rows: (input.markRows ?? [{ outbox_event_id: "outbox-001" }]) as Row[], }; }, }; diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts index 9234b21277..ad0ca9a766 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-outbox-event-source.ts @@ -45,11 +45,15 @@ export function createCockroachOutboxEventSource(input: CreateCockroachOutboxEve })); }, markOutboxEventPublished: async (markInput) => { - await input.executor.execute({ + const result = await input.executor.execute({ name: CockroachOutboxEventSourceStatement.MarkOutboxEventPublished, sql: CockroachOutboxEventSourceSql.MarkOutboxEventPublished, parameters: [markInput.outboxEventId, markInput.publishedAt], }); + + if (result.rows.length !== 1) { + throw new Error(`outbox event was already published or missing: ${markInput.outboxEventId}`); + } }, }; } @@ -59,17 +63,34 @@ type OutboxEventRow = { envelope_json: AgenticEventEnvelope; }; +type PublishedOutboxEventRow = { + outbox_event_id: string; +}; + const CockroachOutboxEventSourceSql = { ClaimUnpublishedOutboxEvents: ` - SELECT outbox_event_id, envelope_json - FROM ${CockroachTableName.OutboxEvents} - WHERE published_at IS NULL - ORDER BY outbox_event_id - LIMIT $1 + UPDATE ${CockroachTableName.OutboxEvents} + SET + claimed_at = transaction_timestamp(), + claim_expires_at = transaction_timestamp() + INTERVAL '5 minutes' + WHERE outbox_event_id IN ( + SELECT outbox_event_id + FROM ${CockroachTableName.OutboxEvents} + WHERE published_at IS NULL + AND (claim_expires_at IS NULL OR claim_expires_at < transaction_timestamp()) + ORDER BY outbox_event_id + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + RETURNING outbox_event_id, envelope_json `, MarkOutboxEventPublished: ` UPDATE ${CockroachTableName.OutboxEvents} - SET published_at = $2 + SET + published_at = $2, + claim_expires_at = NULL WHERE outbox_event_id = $1 + AND published_at IS NULL + RETURNING outbox_event_id `, } as const; diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-schema.test.ts b/agentic-organization/packages/state-cockroach/src/cockroach-schema.test.ts index 7eba2ff572..904162ba15 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-schema.test.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-schema.test.ts @@ -20,6 +20,8 @@ describe("cockroach core state schema", () => { ok(migration.sql.includes("trace_id STRING NOT NULL")); ok(migration.sql.includes("correlation_id STRING NOT NULL")); ok(migration.sql.includes("envelope_json JSONB NOT NULL")); + ok(migration.sql.includes("claimed_at TIMESTAMPTZ")); + ok(migration.sql.includes("claim_expires_at TIMESTAMPTZ")); ok(migration.sql.includes("result_json JSONB NOT NULL")); }); }); diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-schema.ts b/agentic-organization/packages/state-cockroach/src/cockroach-schema.ts index 4d88de366a..b1e0381d8d 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-schema.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-schema.ts @@ -93,6 +93,8 @@ CREATE TABLE IF NOT EXISTS ${CockroachTableName.OutboxEvents} ( trace_id STRING NOT NULL, correlation_id STRING NOT NULL, envelope_json JSONB NOT NULL, + claimed_at TIMESTAMPTZ, + claim_expires_at TIMESTAMPTZ, published_at TIMESTAMPTZ );`.trim(); }