From 21f02990b5e528c32969a220b48af11174e2695d Mon Sep 17 00:00:00 2001 From: "maximka.dolphin" Date: Sun, 21 Jun 2026 14:18:02 +0000 Subject: [PATCH 1/2] test(agentic-org): add KIND e2e supervisor-signal flow driver Publishes one supervisor_signal.sent envelope onto the JetStream stream the worker consumes, to exercise the documented observe -> legal menu -> model decides -> legality re-check loop end to end against the in-cluster substrate (Cockroach + NATS + Ollama). Mirrors deploy/provision-nats.ts (host-run against a port-forwarded NATS). --- .../deploy/drive-supervisor-signal.ts | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 agentic-organization/deploy/drive-supervisor-signal.ts diff --git a/agentic-organization/deploy/drive-supervisor-signal.ts b/agentic-organization/deploy/drive-supervisor-signal.ts new file mode 100644 index 0000000000..b9dbd71fec --- /dev/null +++ b/agentic-organization/deploy/drive-supervisor-signal.ts @@ -0,0 +1,105 @@ +/** + * Drive one supervisor-signal flow end to end against the in-cluster substrate. + * + * Publishes a single `supervisor_signal.sent` AgenticEventEnvelope onto the + * JetStream stream the worker consumes. The worker's V0 automation rules turn it + * into a `CreateSupervisorTriage` reaction plan, executes it through the Hermes + * agent run, and the agent's model-backed composer asks the in-cluster Ollama + * model which legal move to make (the decision kernel re-checks the choice, so + * the model can never widen the rules). This is the loop the docs describe: + * observe -> legal menu -> model decides -> legality re-check. + * + * Run from the host against a port-forwarded NATS (mirrors provision-nats.ts): + * + * kubectl -n agentic-org port-forward svc/nats 4222:4222 & + * node --experimental-strip-types deploy/drive-supervisor-signal.ts + * + * Config (env overrides): + * NATS_PROVISION_SERVERS (default nats://127.0.0.1:4222) + * NATS_PROVISION_ENV (default dev) + * NATS_PROVISION_ORG (default org-lfg) + */ + +import { randomUUID } from "node:crypto"; +import { env } from "node:process"; + +import { connect } from "@nats-io/transport-node"; +import { jetstream } from "@nats-io/jetstream"; + +import { + AgenticAggregateType, + AgenticEventType, + EventSchemaVersion, + SupervisorChainLevel, + createAgenticEventEnvelope, +} from "../packages/domain/src/index.ts"; +import { AgenticMessagingDomain, buildAgenticEventSubject } from "../packages/messaging/src/index.ts"; + +const servers = env.NATS_PROVISION_SERVERS ?? "nats://127.0.0.1:4222"; +const environment = env.NATS_PROVISION_ENV ?? "dev"; +const organizationId = env.NATS_PROVISION_ORG ?? "org-lfg"; + +async function main(): Promise { + const runId = randomUUID(); + const subject = buildAgenticEventSubject({ + environment, + organizationId, + domain: AgenticMessagingDomain.SupervisorSignal, + eventType: AgenticEventType.SupervisorSignalSent, + }); + + const envelope = createAgenticEventEnvelope({ + eventId: `evt-${runId}`, + eventType: AgenticEventType.SupervisorSignalSent, + schemaVersion: EventSchemaVersion.AgenticOrgEventV1, + occurredAt: new Date().toISOString(), + actor: { + agentId: `agent-driver-${runId}`, + hatAssignmentId: `hat-assignment-driver-${runId}`, + }, + scope: { + organizationId, + projectId: `project-${runId}`, + teamId: `team-${runId}`, + workItemId: `work-item-${runId}`, + }, + aggregate: { + aggregateId: `supervisor-signal-${runId}`, + aggregateType: AgenticAggregateType.SupervisorSignal, + aggregateVersion: 1, + }, + trace: { + commandId: `command-${runId}`, + correlationId: `correlation-${runId}`, + causationId: `causation-${runId}`, + traceId: `trace-${runId}`, + idempotencyKey: `idempotency-${runId}`, + }, + payload: { + targetHatAssignmentId: `hat-assignment-target-${runId}`, + targetLevel: SupervisorChainLevel.Manager, + }, + }); + + const connection = await connect({ servers }); + try { + const js = jetstream(connection); + const ack = await js.publish(subject, JSON.stringify(envelope)); + console.log( + JSON.stringify({ + published: true, + subject, + eventId: envelope.eventId, + stream: ack.stream, + seq: ack.seq, + }), + ); + } finally { + await connection.drain(); + } +} + +void main().catch((error: unknown) => { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; +}); From ce0bfe1efbe5891e947af931066129e4fccae783 Mon Sep 17 00:00:00 2001 From: "maximka.dolphin" Date: Sun, 21 Jun 2026 14:41:10 +0000 Subject: [PATCH 2/2] fix(agentic-org): authorize synthesized reaction actors + gate loops on schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the three flow bugs surfaced by the KIND e2e supervisor-signal driver. Bug #1 — synthesized reaction actor had no hat authority, so every org command (e.g. CreateDiscussionAnchor) was denied with hat_authority_missing. The authority projection (agentic_org_hat_assignment_authorities) had a reader but no writer, so nothing populated it for the synthetic agent-reaction-. - Add HatAssignmentAuthorityWriterPort + HatAssignmentAuthorityGrant (ports.ts) - Add createCockroachHatAssignmentAuthorityWriter (idempotent UPSERT) and wire it into the durable state adapters. - In the reaction substrate seeder, grant the synthetic actor its scoped, auditable authority FIRST and unconditionally (before the work-item check), under a real hat (ReactionActorHatId) whose tool bundles cover the action. Bug #2 — re-traced toolTypeForReactionAction: EngineeringManager maps to Prioritize (BacklogAndDefect), which engineering_manager already carries. The supervisor-triage path was already authorized; no fix needed. A mapping-validity test now guards every RequiredHat -> hat -> required-bundle so a future drift can't make a grant vacuous. Bug #3 — the always-on keep-alive / org-cadence loops are started in main() outside the worker process, so their first tick raced the schema migration (which only ran inside the work loop's bootstrap phase) and threw on cold start. Run the migration bootstrap explicitly once, before any loop starts; the worker process takes no bootstrappers (migration is no longer duplicated). Tests: - organization-executor-composition: self-grant path proven without the old stub (grant precedes authorization read); mapping validity. - cockroach-hat-assignment-authority-writer: UPSERT round-trips through reader. - main: loops never tick when the migration bootstrap fails (cold-start gate). --- agentic-organization/apps/workers/src/main.ts | 24 +++- .../src/organization-executor-composition.ts | 51 +++++++- .../apps/workers/test/main.test.ts | 40 +++++++ .../organization-executor-composition.test.ts | 113 ++++++++++++++---- .../packages/application/src/index.ts | 2 + .../packages/application/src/ports.ts | 12 ++ .../src/cockroach-durable-state-adapters.ts | 6 + ...ckroach-hat-assignment-authority-reader.ts | 5 +- ...ckroach-hat-assignment-authority-writer.ts | 63 ++++++++++ .../packages/state-cockroach/src/index.ts | 5 + ...ch-hat-assignment-authority-writer.test.ts | 93 ++++++++++++++ 11 files changed, 384 insertions(+), 30 deletions(-) create mode 100644 agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-writer.ts create mode 100644 agentic-organization/packages/state-cockroach/test/cockroach-hat-assignment-authority-writer.test.ts diff --git a/agentic-organization/apps/workers/src/main.ts b/agentic-organization/apps/workers/src/main.ts index 94d4c24117..3a5a11c4ff 100644 --- a/agentic-organization/apps/workers/src/main.ts +++ b/agentic-organization/apps/workers/src/main.ts @@ -306,12 +306,18 @@ async function runWorkerWithResolvedConfig(input: RunWorkerWithResolvedConfigInp telemetrySink: runtimePorts.telemetrySink, }, }); + // The durable schema bootstrapper. Run EXPLICITLY below (before any loop) + // rather than via the worker process's bootstrap phase, because the always-on + // keep-alive / org-cadence loops are started here in main — outside the + // process — and must observe a ready schema before their first tick. The + // process therefore takes no bootstrappers; the explicit run gates both the + // always-on loops and the work loop (entrypoint.run, below). + const cockroachMigrationBootstrapper = deps.constructors.createMigrationBootstrapper({ + executor: cockroachExecutor, + }); + const process = createWorkerProcess({ - bootstrappers: [ - deps.constructors.createMigrationBootstrapper({ - executor: cockroachExecutor, - }), - ], + bootstrappers: [], readinessProbes: [ deps.constructors.createReadinessProbe({ client: sqlClient, @@ -335,6 +341,14 @@ async function runWorkerWithResolvedConfig(input: RunWorkerWithResolvedConfigInp maxCycles: deps.maxCycles, }); + // Apply the durable schema BEFORE starting any always-on loop. The keep-alive + // and org-cadence loops below tick immediately and concurrently with the work + // loop (entrypoint.run, further down). Their first tick reads/writes durable + // state, so without this gate tick 1 races the migration and throws, + // degrading the cold-start cadence until tick 2. Awaiting it here removes that + // cold-start throw. + await cockroachMigrationBootstrapper.bootstrap(); + // Run the deterministic keep-alive on its own cadence, concurrently with the // work loop. The org heartbeat ticks every LoopIntervalMs regardless of what // the work loop is doing (a slow agent run or a 30s idle NATS poll can no diff --git a/agentic-organization/apps/workers/src/organization-executor-composition.ts b/agentic-organization/apps/workers/src/organization-executor-composition.ts index a808c795c1..9e3af7e1c1 100644 --- a/agentic-organization/apps/workers/src/organization-executor-composition.ts +++ b/agentic-organization/apps/workers/src/organization-executor-composition.ts @@ -7,7 +7,9 @@ */ import { + HatAssignmentAuthorityState, ProjectStatus, + RequiredHat, WorkItemState, WorkItemType, type AgenticActor, @@ -24,6 +26,7 @@ import { createOrganizationReactionPlanActionExecutor, type CommandResult, type EnsureWorkItemPort, + type HatAssignmentAuthorityWriterPort, } from "../../../packages/application/src/index.ts"; import { createCommandAuthorizationPort, createPolicyDecisionObservationPort } from "../../../packages/policy/src/index.ts"; import { @@ -69,8 +72,9 @@ export function composeOrganizationReactionPlanActionExecutor( createId: input.createId, }); - const ensureWorkItem = createCockroachWorkItemSeeder({ + const ensureWorkItem = createReactionSubstrateSeeder({ store: stateAdapters.workAnchorStateStore, + authorityWriter: stateAdapters.hatAssignmentAuthorityWriter, now: input.now, createId: input.createId, }); @@ -90,21 +94,62 @@ function synthesizeActor(action: ReactionPlanAction): AgenticActor { }; } +/** + * The real org hat the synthetic reaction actor wears. The reaction plan's + * `requiredHat` is an abstract supervisor LEVEL (RequiredHat.*); the command + * pipeline authorizes against a concrete hat definition whose tool bundles must + * cover the action class the reaction command needs (see `toolTypeForReactionAction`): + * EngineeringManager -> Prioritize -> BacklogAndDefect + * Reviewer -> WriteDoc -> DocumentationContext + * CSuite/Director/ExecutiveBoard -> AssignHat -> HatAuthorization + * Each mapped hat is asserted (in tests) to exist in `buildHatDefinitions()` and + * to carry the required bundle, so the authority grant is never vacuous. + */ +export const ReactionActorHatId: Readonly> = { + [RequiredHat.EngineeringManager]: "engineering_manager", + [RequiredHat.Reviewer]: "readiness_reviewer", + [RequiredHat.CSuite]: "cto", + [RequiredHat.Director]: "engineering_director", + [RequiredHat.ExecutiveBoard]: "executive_board_member", +}; + type WorkAnchorSeederStore = ReturnType>["workAnchorStateStore"]; -function createCockroachWorkItemSeeder(input: { +function createReactionSubstrateSeeder(input: { store: WorkAnchorSeederStore; + authorityWriter: HatAssignmentAuthorityWriterPort; now: () => string; createId: (prefix: string) => string; }): EnsureWorkItemPort { return { ensureWorkItem: async (action: ReactionPlanAction) => { + const actor = synthesizeActor(action); + + // Grant the synthetic reaction actor its scoped, auditable hat authority + // FIRST and unconditionally (idempotent UPSERT) — the org command that + // follows is gated by the hat-authority policy, and nothing else populates + // the authority projection for a synthesized actor. Done outside the + // work-item existence check so a pre-seeded work item still gets authority. + await input.authorityWriter.grantHatAssignmentAuthority({ + hatAssignmentId: actor.hatAssignmentId, + hatId: ReactionActorHatId[action.requiredHat], + organizationId: action.organizationId, + projectId: action.projectId, + ...(action.teamId === undefined ? {} : { teamId: action.teamId }), + assignedAgentId: actor.agentId, + state: HatAssignmentAuthorityState.Active, + updatedAt: input.now(), + version: 1, + correlationId: action.triggerEventId, + causationId: action.triggerEventId, + traceId: action.triggerEventId, + }); + const existing = await input.store.findWorkItem(action.workItemId); if (existing !== undefined) { return; } - const actor = synthesizeActor(action); const ts = input.now(); const metadata = { updatedAt: ts, diff --git a/agentic-organization/apps/workers/test/main.test.ts b/agentic-organization/apps/workers/test/main.test.ts index 14dfdf3de1..9bb4761f9a 100644 --- a/agentic-organization/apps/workers/test/main.test.ts +++ b/agentic-organization/apps/workers/test/main.test.ts @@ -133,6 +133,46 @@ describe("worker main composition entrypoint", () => { equal(exitCode, WorkerMainTestExitCode.Success); equal(natsAdapters.telemetryWasProvided, true); }); + + test("gates the always-on loops behind durable schema readiness (no cold-start tick before migration)", async () => { + // If the migration bootstrap fails, the keep-alive / org-cadence loops must + // never have started — proving they tick only AFTER the schema is ready and + // can no longer race the migration on a cold start. + const logger = createRecordingLogger(); + const failingBootstrap: RecordingBootstrap = { + bootstrapCount: 0, + bootstrap: async () => { + throw new Error("migration not ready"); + }, + }; + + const exitCode = await runMain( + createTestDependencies({ + logger, + signalRegistrar: createRecordingSignalRegistrar(), + clock: createDeterministicClock(), + shutdownPool: createRecordingShutdownPool(), + natsAdapters: createRecordingNatsAdapters(), + bootstrap: failingBootstrap, + }), + ); + + equal(exitCode, WorkerMainTestExitCode.Degraded); + ok( + logger.records.some( + (record) => + record.stream === WorkerMainLogStream.Stderr && record.message.includes("worker run failed"), + ), + ); + ok( + !logger.records.some((record) => record.message.includes("keep_alive.tick")), + "keep-alive loop must not tick before the schema is ready", + ); + ok( + !logger.records.some((record) => record.message.includes("org_cadence.tick")), + "org-cadence loops must not tick before the schema is ready", + ); + }); }); type CreateTestDependenciesInput = { diff --git a/agentic-organization/apps/workers/test/organization-executor-composition.test.ts b/agentic-organization/apps/workers/test/organization-executor-composition.test.ts index 672ba2b994..70adb6bea2 100644 --- a/agentic-organization/apps/workers/test/organization-executor-composition.test.ts +++ b/agentic-organization/apps/workers/test/organization-executor-composition.test.ts @@ -10,6 +10,11 @@ import { WorkItemType, type ReactionPlanAction, } from "../../../packages/domain/src/index.ts"; +import { + ActionClass, + buildHatDefinitions, + preflightHatAction, +} from "../../../packages/application/src/index.ts"; import { ReactionPlanExecutionStatus, type ReactionPlanActionExecutionContext, @@ -17,11 +22,18 @@ import { type ReactionPlanActionExecutorPort, } from "../../../packages/runtime/src/index.ts"; import type { CockroachOrganizationSqlExecutor } from "../../../packages/state-cockroach/src/index.ts"; -import { composeOrganizationReactionPlanActionExecutor } from "../src/organization-executor-composition.ts"; +import { + ReactionActorHatId, + composeOrganizationReactionPlanActionExecutor, +} from "../src/organization-executor-composition.ts"; describe("organization executor composition", () => { - test("authorizes reaction commands through durable hat assignment authority", async () => { - const cockroachExecutor = createRecordingCockroachExecutor(); + test("self-grants the synthesized reaction actor's hat authority, then authorizes the org command", async () => { + // No authority is pre-seeded — the substrate is empty. The reaction must + // grant its own scoped authority before the command pipeline authorizes the + // discussion-anchor command. (Previously this test stubbed an Active row, + // masking the real-path denial discovered in KIND.) + const cockroachExecutor = createAuthorityStoreCockroachExecutor(); const executor = composeOrganizationReactionPlanActionExecutor({ cockroachExecutor, agentExecutor: createSucceededAgentExecutor(), @@ -32,45 +44,104 @@ describe("organization executor composition", () => { const result = await executor.executeReactionPlanAction(createSupervisorTriageAction(), createExecutionContext()); equal(result.status, ReactionPlanExecutionStatus.Succeeded); - ok(cockroachExecutor.statementNames.includes("find_hat_assignment_authority")); + + const grantIndex = cockroachExecutor.statementNames.indexOf("grant_hat_assignment_authority"); + const findIndex = cockroachExecutor.statementNames.indexOf("find_hat_assignment_authority"); + ok(grantIndex >= 0, "authority is granted"); + ok(findIndex >= 0, "authority is read during authorization"); + ok(grantIndex < findIndex, "authority is granted before it is read"); + + const granted = cockroachExecutor.grantedAuthorities.get("hat-assignment-reaction-engineering_manager"); + ok(granted !== undefined, "synthesized actor has a persisted authority row"); + equal(granted.hat_id, "engineering_manager"); + equal(granted.assigned_agent_id, "agent-reaction-engineering_manager"); + equal(granted.organization_id, "org-lfg"); + equal(granted.project_id, "project-agentic-org"); + equal(granted.team_id, "team-runtime"); + equal(granted.state, "active"); + }); + + test("every reaction-actor hat exists and carries the tool bundle its reaction command needs", () => { + // Guards the ReactionActorHatId mapping against drift: a grant under a hat + // that does not exist (or cannot perform the action class) would be vacuous. + const hatById = new Map(buildHatDefinitions().map((hat) => [hat.id, hat])); + const requiredActionClass: Readonly> = { + [RequiredHat.EngineeringManager]: ActionClass.Prioritize, + [RequiredHat.Reviewer]: ActionClass.WriteDoc, + [RequiredHat.CSuite]: ActionClass.AssignHat, + [RequiredHat.Director]: ActionClass.AssignHat, + [RequiredHat.ExecutiveBoard]: ActionClass.AssignHat, + }; + + for (const requiredHat of Object.values(RequiredHat)) { + const hat = hatById.get(ReactionActorHatId[requiredHat]); + ok(hat !== undefined, `${requiredHat} maps to a real hat (${ReactionActorHatId[requiredHat]})`); + const guardrail = preflightHatAction(hat, requiredActionClass[requiredHat]); + ok(guardrail.allowed, `${hat.id} can perform ${requiredActionClass[requiredHat]} for ${requiredHat}`); + } }); }); -function createRecordingCockroachExecutor(): CockroachOrganizationSqlExecutor & { statementNames: string[] } { +type AuthorityRow = { + hat_assignment_id: string; + hat_id: string; + organization_id: string; + project_id: string; + team_id: string | null; + assigned_agent_id: string; + state: string; +}; + +function createAuthorityStoreCockroachExecutor(): CockroachOrganizationSqlExecutor & { + statementNames: string[]; + grantedAuthorities: Map; +} { const statementNames: string[] = []; + const grantedAuthorities = new Map(); - const executeStatement = async >(statement: { name: string }) => { + const executeStatement = async >(statement: { + name: string; + parameters?: readonly unknown[]; + }) => { statementNames.push(statement.name); + const parameters = statement.parameters ?? []; - if (statement.name === "find_hat_assignment_authority") { - return { - rows: [ - { - hat_assignment_id: "hat-assignment-reaction-engineering_manager", - hat_id: "engineering_manager", - organization_id: "org-lfg", - project_id: "project-agentic-org", - team_id: "team-runtime", - assigned_agent_id: "agent-reaction-engineering_manager", - state: "active", - }, - ] as Row[], + if (statement.name === "grant_hat_assignment_authority") { + const row: AuthorityRow = { + hat_assignment_id: String(parameters[0]), + hat_id: String(parameters[1]), + organization_id: String(parameters[2]), + project_id: String(parameters[3]), + team_id: parameters[4] === null ? null : String(parameters[4]), + assigned_agent_id: String(parameters[5]), + state: String(parameters[6]), }; + grantedAuthorities.set(row.hat_assignment_id, row); + return { rows: [] as Row[] }; } - if (statement.name === "find_work_item") { - return { rows: [workItemRow()] as Row[] }; + if (statement.name === "find_hat_assignment_authority") { + const row = grantedAuthorities.get(String(parameters[0])); + return { rows: (row === undefined ? [] : [row]) as Row[] }; } if (statement.name === "claim_idempotency_record") { return { rows: [{ persistence_status: "committed" }] as Row[] }; } + if (statement.name === "find_work_item") { + // The triage discussion anchor anchors to an existing work item; the grant + // still runs unconditionally (before this existence check) so the + // self-grant path is exercised regardless. + return { rows: [workItemRow()] as Row[] }; + } + return { rows: [] as Row[] }; }; return { statementNames, + grantedAuthorities, execute: executeStatement, executeTransaction: async (operation) => await operation({ execute: executeStatement }), }; diff --git a/agentic-organization/packages/application/src/index.ts b/agentic-organization/packages/application/src/index.ts index a0b3b4042f..f33416dd69 100644 --- a/agentic-organization/packages/application/src/index.ts +++ b/agentic-organization/packages/application/src/index.ts @@ -349,7 +349,9 @@ export type { CommandWorkAnchorWorkItem, ContextPackInboxAnchorStateReaderPort, DiscussionAnchorStateReaderPort, + HatAssignmentAuthorityGrant, HatAssignmentAuthorityReaderPort, + HatAssignmentAuthorityWriterPort, IdGenerator, QualityGateEvaluationStateReaderPort, QualityGateEvaluationWorkItemLookup, diff --git a/agentic-organization/packages/application/src/ports.ts b/agentic-organization/packages/application/src/ports.ts index a58c149158..8f81b2797d 100644 --- a/agentic-organization/packages/application/src/ports.ts +++ b/agentic-organization/packages/application/src/ports.ts @@ -110,6 +110,18 @@ export type HatAssignmentAuthorityReaderPort = { ) => Promise; }; +export type HatAssignmentAuthorityGrant = HatAssignmentAuthoritySnapshot & { + updatedAt: string; + version: number; + correlationId: string; + causationId: string; + traceId: string; +}; + +export type HatAssignmentAuthorityWriterPort = { + grantHatAssignmentAuthority: (grant: HatAssignmentAuthorityGrant) => Promise; +}; + export type SupervisorSignalStateReaderPort = { findSupervisorSignal: (supervisorSignalId: string) => Promise; }; diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-durable-state-adapters.ts b/agentic-organization/packages/state-cockroach/src/cockroach-durable-state-adapters.ts index 9506eea601..598a03f0b4 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-durable-state-adapters.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-durable-state-adapters.ts @@ -1,6 +1,7 @@ import type { CommandStateStoreFactory, HatAssignmentAuthorityReaderPort, + HatAssignmentAuthorityWriterPort, QualityGateEvaluationStateReaderPort, WorkScheduleBlockAuthorityReaderPort, } from "../../application/src/ports.ts"; @@ -22,6 +23,7 @@ import { createCockroachHatAssignmentAuthorityReader, type CockroachHatAssignmentAuthoritySqlExecutor, } from "./cockroach-hat-assignment-authority-reader.ts"; +import { createCockroachHatAssignmentAuthorityWriter } from "./cockroach-hat-assignment-authority-writer.ts"; import { createCockroachOutboxEventSource, type CockroachOutboxSqlExecutor } from "./cockroach-outbox-event-source.ts"; import { createCockroachPolicyDecisionObservationStore, @@ -66,6 +68,7 @@ export type CockroachDurableStateAdapters = { qualityGateEvaluationStateReader: QualityGateEvaluationStateReaderPort; discussionAnchorStateReader: DiscussionAnchorStateReaderPort; hatAssignmentAuthorityReader: HatAssignmentAuthorityReaderPort; + hatAssignmentAuthorityWriter: HatAssignmentAuthorityWriterPort; reactionPlanWorkQueue: ReactionPlanWorkQueue; workScheduleBlockAuthorityReader: WorkScheduleBlockAuthorityReaderPort; workAnchorStateStore: WorkAnchorStateStore; @@ -100,6 +103,9 @@ export function createCockroachDurableStateAdapters( hatAssignmentAuthorityReader: createCockroachHatAssignmentAuthorityReader({ executor: input.executor, }), + hatAssignmentAuthorityWriter: createCockroachHatAssignmentAuthorityWriter({ + executor: input.executor, + }), reactionPlanWorkQueue: createCockroachReactionPlanWorkQueue({ executor: input.executor, }), diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-reader.ts b/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-reader.ts index 5fea6e90de..98d56ec18a 100644 --- a/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-reader.ts +++ b/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-reader.ts @@ -13,7 +13,10 @@ export type CockroachHatAssignmentAuthorityReaderStatement = (typeof CockroachHatAssignmentAuthorityReaderStatement)[keyof typeof CockroachHatAssignmentAuthorityReaderStatement]; export type CockroachHatAssignmentAuthoritySqlStatement = { - name: CockroachHatAssignmentAuthorityReaderStatement; + // `string` (not the reader-only statement enum) so the same executor type is + // shared by the reader and the sibling authority WRITER without a circular + // import; this mirrors the underlying generic Cockroach executor (name: string). + name: string; sql: string; parameters: readonly unknown[]; }; diff --git a/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-writer.ts b/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-writer.ts new file mode 100644 index 0000000000..190d74abd1 --- /dev/null +++ b/agentic-organization/packages/state-cockroach/src/cockroach-hat-assignment-authority-writer.ts @@ -0,0 +1,63 @@ +import type { + HatAssignmentAuthorityGrant, + HatAssignmentAuthorityWriterPort, +} from "../../application/src/index.ts"; +import { CockroachTableName } from "./cockroach-schema.ts"; +import type { CockroachHatAssignmentAuthoritySqlExecutor } from "./cockroach-hat-assignment-authority-reader.ts"; + +export const CockroachHatAssignmentAuthorityWriterStatement = { + GrantHatAssignmentAuthority: "grant_hat_assignment_authority", +} as const; + +export type CockroachHatAssignmentAuthorityWriterStatement = + (typeof CockroachHatAssignmentAuthorityWriterStatement)[keyof typeof CockroachHatAssignmentAuthorityWriterStatement]; + +export type CreateCockroachHatAssignmentAuthorityWriterInput = { + executor: CockroachHatAssignmentAuthoritySqlExecutor; +}; + +export function createCockroachHatAssignmentAuthorityWriter( + input: CreateCockroachHatAssignmentAuthorityWriterInput, +): HatAssignmentAuthorityWriterPort { + return { + grantHatAssignmentAuthority: async (grant: HatAssignmentAuthorityGrant) => { + await input.executor.execute({ + name: CockroachHatAssignmentAuthorityWriterStatement.GrantHatAssignmentAuthority, + sql: CockroachHatAssignmentAuthorityWriterSql.GrantHatAssignmentAuthority, + parameters: [ + grant.hatAssignmentId, + grant.hatId, + grant.organizationId, + grant.projectId, + grant.teamId ?? null, + grant.assignedAgentId, + grant.state, + grant.updatedAt, + grant.version, + grant.correlationId, + grant.causationId, + grant.traceId, + ], + }); + }, + }; +} + +const CockroachHatAssignmentAuthorityWriterSql = { + GrantHatAssignmentAuthority: ` + UPSERT INTO ${CockroachTableName.HatAssignmentAuthorities} ( + hat_assignment_id, + hat_id, + organization_id, + project_id, + team_id, + assigned_agent_id, + state, + updated_at, + version, + correlation_id, + causation_id, + trace_id + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + `, +} as const; diff --git a/agentic-organization/packages/state-cockroach/src/index.ts b/agentic-organization/packages/state-cockroach/src/index.ts index 09a8dd6a3a..3d36e9408c 100644 --- a/agentic-organization/packages/state-cockroach/src/index.ts +++ b/agentic-organization/packages/state-cockroach/src/index.ts @@ -95,6 +95,11 @@ export { type CockroachHatAssignmentAuthoritySqlStatement, type CreateCockroachHatAssignmentAuthorityReaderInput, } from "./cockroach-hat-assignment-authority-reader.ts"; +export { + CockroachHatAssignmentAuthorityWriterStatement, + createCockroachHatAssignmentAuthorityWriter, + type CreateCockroachHatAssignmentAuthorityWriterInput, +} from "./cockroach-hat-assignment-authority-writer.ts"; export { CockroachPolicyDecisionObservationStoreStatement, createCockroachPolicyDecisionObservationStore, diff --git a/agentic-organization/packages/state-cockroach/test/cockroach-hat-assignment-authority-writer.test.ts b/agentic-organization/packages/state-cockroach/test/cockroach-hat-assignment-authority-writer.test.ts new file mode 100644 index 0000000000..0260c58cd5 --- /dev/null +++ b/agentic-organization/packages/state-cockroach/test/cockroach-hat-assignment-authority-writer.test.ts @@ -0,0 +1,93 @@ +import { deepEqual, equal } from "node:assert/strict"; +import { describe, test } from "node:test"; + +import { HatAssignmentAuthorityState } from "../../domain/src/index.ts"; +import type { HatAssignmentAuthorityGrant } from "../../application/src/index.ts"; +import { + CockroachHatAssignmentAuthorityWriterStatement, + createCockroachHatAssignmentAuthorityReader, + createCockroachHatAssignmentAuthorityWriter, + type CockroachHatAssignmentAuthoritySqlExecutor, + type CockroachHatAssignmentAuthoritySqlStatement, +} from "../src/index.ts"; + +describe("cockroach hat assignment authority writer", () => { + test("upserts a grant the reader then resolves as Active behind the generic ports", async () => { + const executor = createInMemoryAuthorityExecutor(); + const writer = createCockroachHatAssignmentAuthorityWriter({ executor }); + const reader = createCockroachHatAssignmentAuthorityReader({ executor }); + + await writer.grantHatAssignmentAuthority(grant()); + + equal(executor.statements[0]?.name, CockroachHatAssignmentAuthorityWriterStatement.GrantHatAssignmentAuthority); + deepEqual(await reader.findHatAssignmentAuthority("hat-assignment-reaction-engineering_manager"), { + hatAssignmentId: "hat-assignment-reaction-engineering_manager", + hatId: "engineering_manager", + organizationId: "org-lfg", + projectId: "project-agentic-org", + teamId: "team-runtime", + assignedAgentId: "agent-reaction-engineering_manager", + state: HatAssignmentAuthorityState.Active, + }); + }); + + test("persists a null team scope for org/project-scoped grants", async () => { + const executor = createInMemoryAuthorityExecutor(); + const writer = createCockroachHatAssignmentAuthorityWriter({ executor }); + const reader = createCockroachHatAssignmentAuthorityReader({ executor }); + + const { teamId: _omitted, ...orgScopedGrant } = grant(); + await writer.grantHatAssignmentAuthority(orgScopedGrant); + + const stored = await reader.findHatAssignmentAuthority("hat-assignment-reaction-engineering_manager"); + equal(stored?.teamId, undefined); + }); +}); + +function grant(): HatAssignmentAuthorityGrant { + return { + hatAssignmentId: "hat-assignment-reaction-engineering_manager", + hatId: "engineering_manager", + organizationId: "org-lfg", + projectId: "project-agentic-org", + teamId: "team-runtime", + assignedAgentId: "agent-reaction-engineering_manager", + state: HatAssignmentAuthorityState.Active, + updatedAt: "2026-05-30T00:00:00.000Z", + version: 1, + correlationId: "evt-supervisor-signal-001", + causationId: "evt-supervisor-signal-001", + traceId: "evt-supervisor-signal-001", + }; +} + +function createInMemoryAuthorityExecutor(): CockroachHatAssignmentAuthoritySqlExecutor & { + statements: CockroachHatAssignmentAuthoritySqlStatement[]; +} { + const statements: CockroachHatAssignmentAuthoritySqlStatement[] = []; + const rows = new Map>(); + + return { + statements, + execute: async >(statement: CockroachHatAssignmentAuthoritySqlStatement) => { + statements.push(statement); + const parameters = statement.parameters; + + if (statement.name === CockroachHatAssignmentAuthorityWriterStatement.GrantHatAssignmentAuthority) { + rows.set(String(parameters[0]), { + hat_assignment_id: parameters[0], + hat_id: parameters[1], + organization_id: parameters[2], + project_id: parameters[3], + team_id: parameters[4], + assigned_agent_id: parameters[5], + state: parameters[6], + }); + return { rows: [] as readonly Row[] }; + } + + const row = rows.get(String(parameters[0])); + return { rows: (row === undefined ? [] : [row]) as readonly unknown[] as readonly Row[] }; + }, + }; +}