feat(daemon): add durable target control plane

Author: cat9999aaa

apps/daemon/package.json

@@ -10,6 +10,7 @@
   },
   "dependencies": {
     "@siteops/ai-proposals": "workspace:*",
+    "@siteops/baota-adapter": "workspace:*",
     "@siteops/core": "workspace:*",
     "@siteops/release-manifest": "workspace:*",
     "@siteops/site-graph": "workspace:*",

apps/daemon/src/db/migrations/005_target_control_plane.sql

@@ -0,0 +1,27 @@
+CREATE TABLE target_profiles (
+  target_id TEXT PRIMARY KEY REFERENCES baota_targets(id) ON DELETE CASCADE,
+  source TEXT NOT NULL CHECK (source IN ('linux', 'simulator')),
+  group_labels_json TEXT NOT NULL CHECK (json_valid(group_labels_json)),
+  snapshot_id TEXT REFERENCES capability_snapshots(id),
+  signed_snapshot_json TEXT CHECK (
+    signed_snapshot_json IS NULL OR json_valid(signed_snapshot_json)
+  ),
+  snapshot_path TEXT,
+  last_checked_at TEXT
+) STRICT;
+
+CREATE TABLE target_inventory_records (
+  target_id TEXT NOT NULL REFERENCES baota_targets(id) ON DELETE CASCADE,
+  resource_id TEXT NOT NULL,
+  resource_kind TEXT NOT NULL,
+  fingerprint TEXT NOT NULL,
+  lifecycle TEXT NOT NULL CHECK (
+    lifecycle IN ('active', 'stale', 'deleted')
+  ),
+  generation INTEGER NOT NULL CHECK (generation > 0),
+  observed_at TEXT NOT NULL,
+  PRIMARY KEY (target_id, resource_id)
+) STRICT;
+
+CREATE INDEX target_inventory_lifecycle_idx
+  ON target_inventory_records(target_id, lifecycle, resource_id);

apps/daemon/src/db/migrations/006_target_job_execution.sql

@@ -0,0 +1,25 @@
+ALTER TABLE operations ADD COLUMN impact_digest TEXT NOT NULL
+  DEFAULT 'sha256:0000000000000000000000000000000000000000000000000000000000000000';
+ALTER TABLE operations ADD COLUMN actor_channel TEXT NOT NULL DEFAULT 'cli'
+  CHECK (actor_channel IN ('console', 'cli', 'mcp'));
+ALTER TABLE operations ADD COLUMN payload_json TEXT
+  CHECK (payload_json IS NULL OR json_valid(payload_json));
+
+ALTER TABLE approvals ADD COLUMN capability TEXT NOT NULL DEFAULT '';
+ALTER TABLE approvals ADD COLUMN operation_class TEXT NOT NULL DEFAULT 'reversible'
+  CHECK (operation_class IN (
+    'read_only',
+    'reversible',
+    'compensatable',
+    'snapshot_restorable',
+    'irreversible'
+  ));
+ALTER TABLE approvals ADD COLUMN impact_digest TEXT NOT NULL
+  DEFAULT 'sha256:0000000000000000000000000000000000000000000000000000000000000000';
+ALTER TABLE approvals ADD COLUMN state TEXT NOT NULL DEFAULT 'active'
+  CHECK (state IN ('active', 'consumed', 'cancelled'));
+ALTER TABLE approvals ADD COLUMN typed_acknowledgement TEXT;
+
+CREATE UNIQUE INDEX jobs_consumed_approval_uq
+  ON jobs(approval_id)
+  WHERE approval_id IS NOT NULL AND deleted_at IS NULL;

apps/daemon/src/db/migrations/__test__/migrations.test.ts

@@ -172,7 +172,7 @@ describe("SQLite control-plane migrations", () => {
     // Given a database stamped with a schema version newer than this runner
     const database = new DatabaseSync(":memory:");
     openDatabases.push(database);
-    database.exec("PRAGMA user_version = 5");
+    database.exec("PRAGMA user_version = 7");
 
     // When this migration runner attempts to open it
     const migrateStaleRunner = () => {
@@ -195,7 +195,7 @@ describe("SQLite control-plane migrations", () => {
       foundVersion: migrationError?.foundVersion,
     }).toEqual({
       name: "MigrationStateError",
-      foundVersion: 5,
+      foundVersion: 7,
     });
   });
 });

apps/daemon/src/db/migrations/index.ts

@@ -1,7 +1,7 @@
 import { readFileSync } from "node:fs";
 import type { DatabaseSync, SQLInputValue } from "node:sqlite";
 
-const CURRENT_SCHEMA_VERSION = 4;
+const CURRENT_SCHEMA_VERSION = 6;
 const CONTROL_PLANE_SQL = readFileSync(
   new URL("./001_control_plane.sql", import.meta.url),
   "utf8",
@@ -18,11 +18,21 @@ const RELEASE_ORCHESTRATION_SQL = readFileSync(
   new URL("./004_release_orchestration.sql", import.meta.url),
   "utf8",
 );
+const TARGET_CONTROL_PLANE_SQL = readFileSync(
+  new URL("./005_target_control_plane.sql", import.meta.url),
+  "utf8",
+);
+const TARGET_JOB_EXECUTION_SQL = readFileSync(
+  new URL("./006_target_job_execution.sql", import.meta.url),
+  "utf8",
+);
 const MIGRATIONS = [
   { version: 1, sql: CONTROL_PLANE_SQL },
   { version: 2, sql: VAULT_SQL },
   { version: 3, sql: DURABLE_JOBS_SQL },
   { version: 4, sql: RELEASE_ORCHESTRATION_SQL },
+  { version: 5, sql: TARGET_CONTROL_PLANE_SQL },
+  { version: 6, sql: TARGET_JOB_EXECUTION_SQL },
 ] as const;
 
 export type ConstraintKind = "foreign_key" | "unique" | "check" | "other";

apps/daemon/src/http/contracts.ts

@@ -18,6 +18,21 @@ const ResourceKindSchema = z.enum([
   "runtime_service",
   "data_store",
 ]);
+export const OperationPayloadSchema = z.discriminatedUnion("kind", [
+  z
+    .object({ kind: z.literal("inventory.read") })
+    .strict()
+    .readonly(),
+  z
+    .object({
+      kind: z.literal("site.settings.update"),
+      siteId: z.string().min(1),
+      priorNote: z.string(),
+      note: z.string(),
+    })
+    .strict()
+    .readonly(),
+]);
 
 export const DraftInputSchema = z
   .object({
@@ -37,6 +52,7 @@ export const DraftInputSchema = z
       })
       .strict()
       .readonly(),
+    payload: OperationPayloadSchema.nullable().default(null),
   })
   .strict()
   .readonly();

apps/daemon/src/http/execute-route.ts

@@ -102,6 +102,13 @@ export function registerExecuteRoute(
       timeoutAt: input.data.timeoutAt,
       now: options.now(),
     });
+    if (decision.approvalId !== null) {
+      options.database
+        .prepare(
+          "UPDATE approvals SET state = 'consumed', updated_at = ? WHERE id = ? AND state = 'active'",
+        )
+        .run(options.now(), decision.approvalId);
+    }
     options.audit.append({
       actorId: actor.id,
       action: "job:queued",

apps/daemon/src/http/index.ts

@@ -13,12 +13,14 @@ import {
   type SyntheticProbe,
 } from "../observability/synthetic.js";
 import { TelemetryStore } from "../observability/store.js";
+import type { DurableTargetControlPlane } from "../targets/control-plane.js";
 import { AuditStore } from "./audit.js";
 import { registerJobRoutes } from "./jobs-routes.js";
 import { registerIntakeRoute } from "./intake-route.js";
 import { registerObservabilityRoutes } from "./observability-routes.js";
 import { registerReadRoutes } from "./read-routes.js";
 import { loopbackGuard } from "./security.js";
+import { registerTargetRoutes } from "./target-routes.js";
 
 export interface DaemonHttpOptions {
   readonly database: DatabaseSync;
@@ -28,6 +30,7 @@ export interface DaemonHttpOptions {
   readonly intake: IntakeService;
   readonly telemetryStore?: TelemetryStore;
   readonly syntheticProbe?: SyntheticProbe;
+  readonly targets?: DurableTargetControlPlane;
 }
 
 export function buildDaemonHttp(options: DaemonHttpOptions): FastifyInstance {
@@ -77,6 +80,13 @@ export function buildDaemonHttp(options: DaemonHttpOptions): FastifyInstance {
     store: telemetryStore,
     synthetic,
   });
+  if (options.targets !== undefined) {
+    registerTargetRoutes(app, {
+      auth: options.auth,
+      audit,
+      targets: options.targets,
+    });
+  }
   return app;
 }
 

apps/daemon/src/http/jobs-routes.ts

@@ -7,6 +7,7 @@ import {
   type BoundApproval,
   type PolicyOperation,
 } from "@siteops/core";
+import type { z } from "zod";
 
 import { ApprovalError, ApprovalService } from "../approvals/index.js";
 import { CAPABILITIES, type LocalAuth } from "../auth/index.js";
@@ -16,13 +17,15 @@ import {
   ApprovalInputSchema,
   DraftInputSchema,
   IdParamsSchema,
+  OperationPayloadSchema,
 } from "./contracts.js";
 import { registerExecuteRoute } from "./execute-route.js";
 import { loadPolicyContext } from "./policy-context.js";
 import { authorizeRequest } from "./security.js";
 
 export interface DraftRecord {
   readonly operation: PolicyOperation;
+  readonly payload: z.infer<typeof OperationPayloadSchema> | null;
 }
 
 export interface JobRouteOptions {
@@ -97,7 +100,7 @@ export function registerJobRoutes(
     const now = options.now();
     options.database
       .prepare(
-        "INSERT INTO operations (id, schema_version, owner_id, generation, target_id, resource_kind, resource_id, capability, operation_class, payload_digest, state, created_at, updated_at) VALUES (?, 1, ?, 1, ?, ?, ?, ?, ?, ?, 'drafted', ?, ?)",
+        "INSERT INTO operations (id, schema_version, owner_id, generation, target_id, resource_kind, resource_id, capability, operation_class, payload_digest, impact_digest, actor_channel, payload_json, state, created_at, updated_at) VALUES (?, 1, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'drafted', ?, ?)",
       )
       .run(
         operation.id,
@@ -108,10 +111,16 @@ export function registerJobRoutes(
         operation.capability,
         operation.operationClass,
         operation.payloadDigest,
+        operation.declaredImpact.digest,
+        actor.channel,
+        input.data.payload === null ? null : JSON.stringify(input.data.payload),
         now,
         now,
       );
-    drafts.set(operation.id, { operation: context.operation });
+    drafts.set(operation.id, {
+      operation: context.operation,
+      payload: input.data.payload,
+    });
     options.audit.append({
       actorId: actor.id,
       action: "job:drafted",
@@ -147,6 +156,27 @@ export function registerJobRoutes(
         input.data.expiresAt,
         input.data.typedAcknowledgement,
       );
+      options.database
+        .prepare(
+          "INSERT INTO approvals (id, schema_version, owner_id, generation, operation_id, target_id, resource_id, payload_digest, capability, operation_class, impact_digest, approved_by, approved_at, expires_at, state, typed_acknowledgement, created_at, updated_at) VALUES (?, 1, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)",
+        )
+        .run(
+          approval.id,
+          approval.actorId,
+          approval.operationId,
+          approval.targetId,
+          approval.resource.id,
+          approval.payloadDigest,
+          approval.capability,
+          approval.operationClass,
+          approval.impactDigest,
+          actor.id,
+          approval.approvedAt,
+          approval.expiresAt,
+          approval.typedAcknowledgement,
+          approval.approvedAt,
+          approval.approvedAt,
+        );
       approved.set(approval.id, approval);
       options.audit.append({
         actorId: actor.id,

apps/daemon/src/http/target-routes.ts

@@ -0,0 +1,105 @@
+import type { FastifyInstance } from "fastify";
+import { z } from "zod";
+
+import { CAPABILITIES, type LocalAuth } from "../auth/index.js";
+import {
+  DurableTargetControlPlane,
+  TargetEndpointError,
+  TargetEnrollmentError,
+} from "../targets/control-plane.js";
+import type { AuditStore } from "./audit.js";
+import { IdParamsSchema } from "./contracts.js";
+import { authorizeRequest } from "./security.js";
+
+const EnrollmentBodySchema = z
+  .object({
+    targetId: z.uuid(),
+    name: z.string().min(1),
+    endpoint: z.url(),
+    groupLabels: z.array(z.string().min(1)).readonly(),
+    secret: z.string().min(1).max(65_536),
+    source: z.enum(["linux", "simulator"]),
+  })
+  .strict()
+  .readonly();
+
+interface TargetRouteOptions {
+  readonly auth: LocalAuth;
+  readonly audit: AuditStore;
+  readonly targets: DurableTargetControlPlane;
+}
+
+export function registerTargetRoutes(
+  app: FastifyInstance,
+  options: TargetRouteOptions,
+): void {
+  app.post("/v1/targets", async (request, reply) => {
+    const actor = authorizeRequest(options, {
+      request,
+      reply,
+      capability: CAPABILITIES.jobsPropose,
+    });
+    if (actor === null) {
+      return;
+    }
+    const body = EnrollmentBodySchema.safeParse(request.body);
+    if (!body.success) {
+      await reply.code(400).send({ error: "invalid_request" });
+      return;
+    }
+    try {
+      const target = await options.targets.enroll({
+        ...body.data,
+        ownerId: actor.id,
+      });
+      options.audit.append({
+        actorId: actor.id,
+        action: "target:enrolled",
+        targetId: target.id,
+        resourceKind: "target",
+        resourceId: target.id,
+      });
+      await reply.code(201).send(target);
+    } catch (error) {
+      if (
+        !(error instanceof TargetEndpointError) &&
+        !(error instanceof TargetEnrollmentError)
+      ) {
+        throw error;
+      }
+      options.audit.append({
+        actorId: actor.id,
+        action: `denied:${error.reason}`,
+      });
+      await reply.code(400).send({
+        error: "target_enrollment_rejected",
+        reason: error.reason,
+      });
+    }
+  });
+
+  app.get("/v1/targets/:id", async (request, reply) => {
+    const actor = authorizeRequest(options, {
+      request,
+      reply,
+      capability: CAPABILITIES.daemonRead,
+    });
+    if (actor === null) {
+      return;
+    }
+    const params = IdParamsSchema.safeParse(request.params);
+    if (!params.success) {
+      await reply.code(404).send({ error: "target_not_found" });
+      return;
+    }
+    try {
+      await reply.send(options.targets.read(params.data.id));
+    } catch (error) {
+      if (error instanceof z.ZodError) {
+        await reply.code(404).send({ error: "target_not_found" });
+        return;
+      }
+      throw error;
+    }
+  });
+}