fix(agentic-org): KIND e2e supervisor-signal flow + fix the 3 bugs it surfaced

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

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<Record<RequiredHat, string>> = {
+  [RequiredHat.EngineeringManager]: "engineering_manager",
+  [RequiredHat.Reviewer]: "readiness_reviewer",
+  [RequiredHat.CSuite]: "cto",
+  [RequiredHat.Director]: "engineering_director",
+  [RequiredHat.ExecutiveBoard]: "executive_board_member",
+};
+
 type WorkAnchorSeederStore = ReturnType<typeof createCockroachDurableStateAdapters<CommandResult>>["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,

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 = {

agentic-organization/apps/workers/test/organization-executor-composition.test.ts

@@ -10,18 +10,30 @@ import {
   WorkItemType,
   type ReactionPlanAction,
 } from "../../../packages/domain/src/index.ts";
+import {
+  ActionClass,
+  buildHatDefinitions,
+  preflightHatAction,
+} from "../../../packages/application/src/index.ts";
 import {
   ReactionPlanExecutionStatus,
   type ReactionPlanActionExecutionContext,
   type ReactionPlanActionExecutionResult,
   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<Record<RequiredHat, ActionClass>> = {
+      [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<string, AuthorityRow>;
+} {
   const statementNames: string[] = [];
+  const grantedAuthorities = new Map<string, AuthorityRow>();
 
-  const executeStatement = async <Row = Record<string, unknown>>(statement: { name: string }) => {
+  const executeStatement = async <Row = Record<string, unknown>>(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 }),
   };