feat: 补齐 HWLAB 手动调度能力

src/mgr/kubernetes-runner-job.ts

@@ -1,8 +1,10 @@
 import { spawn } from "node:child_process";
 import { AgentRunError } from "../common/errors.js";
 import { redactJson, redactText } from "../common/redaction.js";
+import { isTerminalCommandState, isTerminalRunStatus, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js";
 import type { AgentRunStore } from "./store.js";
 import type { JsonRecord } from "../common/types.js";
+import { stableHash } from "../common/validation.js";
 import { renderRunnerJobManifest } from "../runner/k8s-job.js";
 
 export interface RunnerJobDefaults {
@@ -23,6 +25,7 @@ export interface CreateRunnerJobInput extends JsonRecord {
   runnerId?: string;
   sourceCommit?: string;
   serviceAccountName?: string;
+  idempotencyKey?: string;
 }
 
 export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults }): Promise<JsonRecord> {
@@ -38,6 +41,15 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
   const managerUrl = optionalString(options.input.managerUrl) ?? options.defaults.managerUrl;
   const sourceCommit = optionalString(options.input.sourceCommit) ?? options.defaults.sourceCommit;
   const serviceAccountName = optionalString(options.input.serviceAccountName) ?? options.defaults.serviceAccountName;
+  const idempotencyKey = optionalString(options.input.idempotencyKey);
+  const normalizedPayload = { commandId, image, namespace, managerUrl, sourceCommit, serviceAccountName: serviceAccountName ?? null, attemptId: optionalString(options.input.attemptId) ?? null, runnerId: optionalString(options.input.runnerId) ?? null };
+  const payloadHash = stableHash(normalizedPayload);
+  if (idempotencyKey) {
+    const existing = await options.store.getRunnerJobByIdempotencyKey(run.id, idempotencyKey, payloadHash);
+    if (existing) return { ...existing.result, idempotentReplay: true };
+  }
+  if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${run.id} is already terminal: ${run.status}`, { httpStatus: 409 });
+  if (isTerminalCommandState(command.state) || command.state !== "pending") throw new AgentRunError(command.state === "cancelled" ? "cancelled" : "schema-invalid", `command ${commandId} is not pending: ${command.state}`, { httpStatus: 409 });
 
   const renderOptions = {
     run,
@@ -52,9 +64,15 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
   const runnerId = optionalString(options.input.runnerId);
   const render = renderRunnerJobManifest({ ...renderOptions, ...(attemptId ? { attemptId } : {}), ...(runnerId ? { runnerId } : {}) });
   const created = await kubectlCreate(render.manifest, options.defaults.kubectlCommand ?? "kubectl");
-  return {
+  const response = {
     action: "create-kubernetes-job",
     mutation: true,
+    runId: run.id,
+    commandId,
+    attemptId: render.attemptId,
+    runnerId: render.runnerId,
+    namespace: render.namespace,
+    jobName: render.jobName,
     jobIdentity: {
       kind: "Job",
       namespace: render.namespace,
@@ -90,7 +108,34 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
       kind: objectPath(created, ["kind"]),
       resourceVersion: objectPath(created, ["metadata", "resourceVersion"]),
     },
-  };
+  } satisfies JsonRecord;
+  const saved = await options.store.saveRunnerJob({
+    runId: run.id,
+    commandId,
+    idempotencyKey: idempotencyKey ?? null,
+    payloadHash,
+    attemptId: render.attemptId,
+    runnerId: render.runnerId,
+    namespace: render.namespace,
+    jobName: render.jobName,
+    managerUrl,
+    image,
+    sourceCommit,
+    serviceAccountName: serviceAccountName ?? null,
+    result: response,
+  });
+  await options.store.appendEvent(run.id, "backend_status", {
+    phase: "runner-job-created",
+    commandId,
+    attemptId: saved.attemptId,
+    runnerId: saved.runnerId,
+    namespace: saved.namespace,
+    jobName: saved.jobName,
+    idempotencyKey: idempotencyKey ? "present" : null,
+    sessionRef: summarizeSessionRef(run.sessionRef ?? null),
+    resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null),
+  });
+  return response;
 }
 
 async function kubectlCreate(manifest: JsonRecord, kubectlCommand: string): Promise<JsonRecord> {

src/mgr/postgres-store.ts

@@ -3,10 +3,10 @@ import { Pool } from "pg";
 import type { PoolClient, QueryResultRow } from "pg";
 import { AgentRunError } from "../common/errors.js";
 import { redactJson } from "../common/redaction.js";
-import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, RunEvent, RunnerRecord, RunRecord, RunStatus, TerminalStatus } from "../common/types.js";
+import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionRecord, SessionRef, TerminalStatus } from "../common/types.js";
 import { newId, nowIso, stableHash } from "../common/validation.js";
-import type { AgentRunStore, StoreHealth } from "./store.js";
-import { commandStateFromTerminal, statusFromTerminal } from "./store.js";
+import type { AgentRunStore, SaveRunnerJobInput, StoreHealth } from "./store.js";
+import { commandStateFromTerminal, isTerminalCommandState, isTerminalRunStatus, sessionRefFromRecord, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js";
 import { backendCapabilitiesSqlValues } from "../common/backend-profiles.js";
 
 interface PostgresStoreOptions {
@@ -126,6 +126,50 @@ ON CONFLICT (profile) DO UPDATE SET
   updated_at = EXCLUDED.updated_at;
 `;
 
+const hwlabManualDispatchMigrationSql = `
+ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS session_ref jsonb;
+ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS resource_bundle_ref jsonb;
+
+CREATE TABLE IF NOT EXISTS agentrun_sessions (
+  session_id text PRIMARY KEY,
+  tenant_id text NOT NULL,
+  project_id text NOT NULL,
+  backend_profile text NOT NULL,
+  conversation_id text,
+  thread_id text,
+  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
+  created_at timestamptz NOT NULL,
+  updated_at timestamptz NOT NULL,
+  expires_at timestamptz
+);
+
+CREATE TABLE IF NOT EXISTS agentrun_runner_jobs (
+  id text PRIMARY KEY,
+  run_id text NOT NULL REFERENCES agentrun_runs(id) ON DELETE CASCADE,
+  command_id text NOT NULL REFERENCES agentrun_commands(id) ON DELETE CASCADE,
+  idempotency_key text,
+  payload_hash text NOT NULL,
+  attempt_id text NOT NULL,
+  runner_id text NOT NULL,
+  namespace text NOT NULL,
+  job_name text NOT NULL,
+  manager_url text NOT NULL,
+  image text NOT NULL,
+  source_commit text NOT NULL,
+  service_account_name text,
+  result jsonb NOT NULL,
+  created_at timestamptz NOT NULL,
+  updated_at timestamptz NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS agentrun_runner_jobs_run_idempotency_key_idx
+  ON agentrun_runner_jobs (run_id, idempotency_key)
+  WHERE idempotency_key IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS agentrun_runner_jobs_run_command_idx ON agentrun_runner_jobs (run_id, command_id, created_at);
+CREATE INDEX IF NOT EXISTS agentrun_sessions_tenant_project_idx ON agentrun_sessions (tenant_id, project_id, backend_profile, updated_at);
+`;
+
 const postgresMigrations: MigrationDefinition[] = [
   {
     id: "001_v01_initial_durable_store",
@@ -137,13 +181,18 @@ const postgresMigrations: MigrationDefinition[] = [
     checksum: checksumSql(backendProfilesMigrationSql),
     sql: backendProfilesMigrationSql,
   },
+  {
+    id: "003_v01_hwlab_manual_dispatch",
+    checksum: checksumSql(hwlabManualDispatchMigrationSql),
+    sql: hwlabManualDispatchMigrationSql,
+  },
 ];
 
 export function postgresMigrationContract(): JsonRecord {
   return {
     migrationIds: postgresMigrations.map((migration) => migration.id),
     latestMigrationId: latestMigrationId(),
-    requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases"],
+    requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases", "agentrun_sessions", "agentrun_runner_jobs"],
     checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])),
   };
 }
@@ -201,14 +250,15 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
 
   async createRun(input: CreateRunInput): Promise<RunRecord> {
     const at = nowIso();
-    const run: RunRecord = { ...input, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null };
     return this.withTransaction(async (client) => {
+      const sessionRef = await this.resolveSessionForRun(client, input, at);
+      const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null };
       await client.query(
-        `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at)
-         VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7::jsonb, $8::jsonb, $9, $10, $11, $12, $13, $14, $15, $16)`,
-        [run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt],
+        `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, session_ref, resource_bundle_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at)
+         VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`,
+        [run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), JSON.stringify(run.sessionRef), JSON.stringify(run.resourceBundleRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt],
       );
-      await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile });
+      await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile, sessionRef: summarizeSessionRef(run.sessionRef ?? null), resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null) });
       return run;
     });
   }
@@ -285,10 +335,55 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return runnerFromRow(result.rows[0]);
   }
 
+  async listRunnerJobs(runId: string, commandId?: string): Promise<RunnerJobRecord[]> {
+    await this.getRun(runId);
+    const params: unknown[] = [runId];
+    let where = "run_id = $1";
+    if (commandId) {
+      params.push(commandId);
+      where += ` AND command_id = $${params.length}`;
+    }
+    const result = await this.pool.query(`SELECT * FROM agentrun_runner_jobs WHERE ${where} ORDER BY created_at ASC`, params);
+    return result.rows.map(runnerJobFromRow);
+  }
+
+  async getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): Promise<RunnerJobRecord | null> {
+    const result = await this.pool.query("SELECT * FROM agentrun_runner_jobs WHERE run_id = $1 AND idempotency_key = $2", [runId, idempotencyKey]);
+    const row = result.rows[0];
+    if (!row) return null;
+    const record = runnerJobFromRow(row);
+    if (record.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "runner job idempotency key reused with different payload", { httpStatus: 409 });
+    return record;
+  }
+
+  async saveRunnerJob(input: SaveRunnerJobInput): Promise<RunnerJobRecord> {
+    return this.withTransaction(async (client) => {
+      await this.requireRunForUpdate(client, input.runId);
+      if (input.idempotencyKey) {
+        const existing = await client.query("SELECT * FROM agentrun_runner_jobs WHERE run_id = $1 AND idempotency_key = $2 FOR UPDATE", [input.runId, input.idempotencyKey]);
+        if (existing.rows[0]) {
+          const record = runnerJobFromRow(existing.rows[0]);
+          if (record.payloadHash !== input.payloadHash) throw new AgentRunError("schema-invalid", "runner job idempotency key reused with different payload", { httpStatus: 409 });
+          return record;
+        }
+      }
+      const at = nowIso();
+      const record: RunnerJobRecord = { ...input, id: newId("rjob"), idempotencyKey: input.idempotencyKey ?? null, serviceAccountName: input.serviceAccountName ?? null, createdAt: at, updatedAt: at };
+      const inserted = await client.query(
+        `INSERT INTO agentrun_runner_jobs (id, run_id, command_id, idempotency_key, payload_hash, attempt_id, runner_id, namespace, job_name, manager_url, image, source_commit, service_account_name, result, created_at, updated_at)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::jsonb, $15, $16)
+         RETURNING *`,
+        [record.id, record.runId, record.commandId, record.idempotencyKey, record.payloadHash, record.attemptId, record.runnerId, record.namespace, record.jobName, record.managerUrl, record.image, record.sourceCommit, record.serviceAccountName, JSON.stringify(record.result), record.createdAt, record.updatedAt],
+      );
+      return runnerJobFromRow(inserted.rows[0]);
+    });
+  }
+
   async claimRun(runId: string, runnerId: string, leaseMs: number): Promise<RunRecord> {
     return this.withTransaction(async (client) => {
       const run = await this.requireRunForUpdate(client, runId);
-      if (run.claimedBy && run.claimedBy !== runnerId && !isTerminalStatus(run.status)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
+      if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 });
+      if (run.claimedBy && run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
       const leaseExpiresAt = new Date(Date.now() + leaseMs).toISOString();
       const updated = await client.query(
         `UPDATE agentrun_runs SET status = $2, claimed_by = $3, lease_expires_at = $4, updated_at = $5 WHERE id = $1 RETURNING *`,
@@ -308,6 +403,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
   async heartbeat(runId: string, runnerId: string, leaseMs: number): Promise<RunRecord> {
     return this.withTransaction(async (client) => {
       const run = await this.requireRunForUpdate(client, runId);
+      if (isTerminalRunStatus(run.status)) return run;
       if (run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is not claimed by ${runnerId}`, { httpStatus: 409 });
       const leaseExpiresAt = new Date(Date.now() + leaseMs).toISOString();
       const updated = await client.query("UPDATE agentrun_runs SET lease_expires_at = $2, updated_at = $3 WHERE id = $1 RETURNING *", [runId, leaseExpiresAt, nowIso()]);
@@ -318,10 +414,15 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
   }
 
   async ackCommand(commandId: string): Promise<CommandRecord> {
-    const result = await this.pool.query("UPDATE agentrun_commands SET state = $2, acknowledged_at = $3, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, "acknowledged", nowIso()]);
-    const row = result.rows[0];
-    if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
-    return commandFromRow(row);
+    return this.withTransaction(async (client) => {
+      const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
+      const row = existing.rows[0];
+      if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
+      const command = commandFromRow(row);
+      if (isTerminalCommandState(command.state) || command.state === "acknowledged") return command;
+      const result = await client.query("UPDATE agentrun_commands SET state = $2, acknowledged_at = $3, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, "acknowledged", nowIso()]);
+      return commandFromRow(result.rows[0]);
+    });
   }
 
   async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): Promise<CommandRecord> {
@@ -330,6 +431,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       const row = existing.rows[0];
       if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
       const command = commandFromRow(row);
+      if (isTerminalCommandState(command.state)) return command;
       const state = commandStateFromTerminal(result.terminalStatus);
       const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, nowIso()]);
       await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
@@ -344,19 +446,67 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     });
   }
 
-  async finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): Promise<RunRecord> {
+  async finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<RunRecord> {
     return this.withTransaction(async (client) => {
-      await this.requireRunForUpdate(client, runId);
+      const existing = await this.requireRunForUpdate(client, runId);
+      if (isTerminalRunStatus(existing.status)) return existing;
       const status = statusFromTerminal(result.terminalStatus);
       const updated = await client.query(
         `UPDATE agentrun_runs SET status = $2, terminal_status = $3, failure_kind = $4, failure_message = $5, updated_at = $6 WHERE id = $1 RETURNING *`,
         [runId, status, result.terminalStatus, result.failureKind, result.failureMessage, nowIso()],
       );
+      const run = runFromRow(updated.rows[0]);
+      if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
       await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage });
+      return run;
+    });
+  }
+
+  async cancelRun(runId: string, reason = "cancel requested"): Promise<RunRecord> {
+    return this.withTransaction(async (client) => {
+      const run = await this.requireRunForUpdate(client, runId);
+      if (isTerminalRunStatus(run.status)) return run;
+      const at = nowIso();
+      const commands = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') FOR UPDATE", [runId]);
+      for (const row of commands.rows) {
+        const command = commandFromRow(row);
+        await client.query("UPDATE agentrun_commands SET state = 'cancelled', updated_at = $2 WHERE id = $1", [command.id, at]);
+        await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
+      }
+      await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "cancel-requested", reason });
+      const updated = await client.query(
+        `UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
+        [runId, reason, at],
+      );
+      await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
       return runFromRow(updated.rows[0]);
     });
   }
 
+  async cancelCommand(commandId: string, reason = "cancel requested"): Promise<CommandRecord> {
+    return this.withTransaction(async (client) => {
+      const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
+      const row = existing.rows[0];
+      if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
+      const command = commandFromRow(row);
+      if (isTerminalCommandState(command.state)) return command;
+      const updated = await client.query("UPDATE agentrun_commands SET state = 'cancelled', updated_at = $2 WHERE id = $1 RETURNING *", [commandId, nowIso()]);
+      await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
+      const run = await this.requireRunForUpdate(client, command.runId);
+      if (!isTerminalRunStatus(run.status)) {
+        await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "cancel-requested", reason });
+        await client.query(`UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, updated_at = $3 WHERE id = $1`, [command.runId, reason, nowIso()]);
+        await this.appendEventWithLockedRun(client, command.runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
+      }
+      return commandFromRow(updated.rows[0]);
+    });
+  }
+
+  async getSession(sessionId: string): Promise<SessionRecord | null> {
+    const result = await this.pool.query("SELECT * FROM agentrun_sessions WHERE session_id = $1", [sessionId]);
+    return result.rows[0] ? sessionFromRow(result.rows[0]) : null;
+  }
+
   async backends(): Promise<JsonRecord[]> {
     const result = await this.pool.query("SELECT * FROM agentrun_backends ORDER BY profile ASC");
     return result.rows.map((row) => ({ profile: stringValue(row.profile), ...jsonRecord(row.capabilities), capacity: jsonValue(row.capacity), health: jsonValue(row.health), updatedAt: nullableIso(row.updated_at) }));
@@ -385,6 +535,67 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return runFromRow(row);
   }
 
+  private async resolveSessionForRun(client: PoolClient, input: CreateRunInput, at: string): Promise<SessionRef | null> {
+    if (!input.sessionRef) return null;
+    const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionRef.sessionId]);
+    if (existing.rows[0]) return sessionRefFromRecord(sessionFromRow(existing.rows[0]), input.sessionRef);
+    const record: SessionRecord = {
+      sessionId: input.sessionRef.sessionId,
+      tenantId: input.tenantId,
+      projectId: input.projectId,
+      backendProfile: input.backendProfile,
+      conversationId: input.sessionRef.conversationId ?? null,
+      threadId: input.sessionRef.threadId ?? null,
+      metadata: input.sessionRef.metadata ?? {},
+      createdAt: at,
+      updatedAt: at,
+      expiresAt: input.sessionRef.expiresAt ?? null,
+    };
+    await client.query(
+      `INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, created_at, updated_at, expires_at)
+       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10)`,
+      [record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.createdAt, record.updatedAt, record.expiresAt],
+    );
+    return sessionRefFromRecord(record, input.sessionRef);
+  }
+
+  private async upsertSessionThread(client: PoolClient, run: RunRecord, threadId: string, turnId: string | null): Promise<void> {
+    if (!run.sessionRef?.sessionId) return;
+    const at = nowIso();
+    const existingResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [run.sessionRef.sessionId]);
+    const existing = existingResult.rows[0] ? sessionFromRow(existingResult.rows[0]) : null;
+    const metadata = { ...(existing?.metadata ?? {}), ...(run.sessionRef.metadata ?? {}), ...(turnId ? { lastTurnId: turnId } : {}) };
+    const record: SessionRecord = {
+      sessionId: run.sessionRef.sessionId,
+      tenantId: run.tenantId,
+      projectId: run.projectId,
+      backendProfile: run.backendProfile,
+      conversationId: run.sessionRef.conversationId ?? existing?.conversationId ?? null,
+      threadId,
+      metadata,
+      createdAt: existing?.createdAt ?? at,
+      updatedAt: at,
+      expiresAt: run.sessionRef.expiresAt ?? existing?.expiresAt ?? null,
+    };
+    await client.query(
+      `INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, created_at, updated_at, expires_at)
+       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10)
+       ON CONFLICT (session_id) DO UPDATE SET
+         tenant_id = EXCLUDED.tenant_id,
+         project_id = EXCLUDED.project_id,
+         backend_profile = EXCLUDED.backend_profile,
+         conversation_id = EXCLUDED.conversation_id,
+         thread_id = EXCLUDED.thread_id,
+         metadata = EXCLUDED.metadata,
+         updated_at = EXCLUDED.updated_at,
+         expires_at = EXCLUDED.expires_at`,
+      [record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.createdAt, record.updatedAt, record.expiresAt],
+    );
+    const nextSessionRef = sessionRefFromRecord(record, run.sessionRef);
+    await client.query("UPDATE agentrun_runs SET session_ref = $2::jsonb, updated_at = $3 WHERE id = $1", [run.id, JSON.stringify(nextSessionRef), at]);
+    await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "session-updated", sessionRef: summarizeSessionRef(nextSessionRef), turnId });
+  }
+
   private async withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
     const client = await this.pool.connect();
     try {
@@ -419,6 +630,8 @@ function runFromRow(row: QueryResultRow): RunRecord {
     tenantId: stringValue(row.tenant_id),
     projectId: stringValue(row.project_id),
     workspaceRef: jsonRecord(row.workspace_ref) as RunRecord["workspaceRef"],
+    sessionRef: jsonValue(row.session_ref) as RunRecord["sessionRef"],
+    resourceBundleRef: jsonValue(row.resource_bundle_ref) as RunRecord["resourceBundleRef"],
     providerId: stringValue(row.provider_id),
     backendProfile: stringValue(row.backend_profile) as BackendProfile,
     executionPolicy: jsonRecord(row.execution_policy) as RunRecord["executionPolicy"],
@@ -468,15 +681,47 @@ function runnerFromRow(row: QueryResultRow): RunnerRecord {
   };
 }
 
+function sessionFromRow(row: QueryResultRow): SessionRecord {
+  return {
+    sessionId: stringValue(row.session_id),
+    tenantId: stringValue(row.tenant_id),
+    projectId: stringValue(row.project_id),
+    backendProfile: stringValue(row.backend_profile) as BackendProfile,
+    conversationId: nullableString(row.conversation_id),
+    threadId: nullableString(row.thread_id),
+    metadata: jsonRecord(row.metadata),
+    createdAt: iso(row.created_at),
+    updatedAt: iso(row.updated_at),
+    expiresAt: nullableIso(row.expires_at),
+  };
+}
+
+function runnerJobFromRow(row: QueryResultRow): RunnerJobRecord {
+  return {
+    id: stringValue(row.id),
+    runId: stringValue(row.run_id),
+    commandId: stringValue(row.command_id),
+    idempotencyKey: nullableString(row.idempotency_key),
+    payloadHash: stringValue(row.payload_hash),
+    attemptId: stringValue(row.attempt_id),
+    runnerId: stringValue(row.runner_id),
+    namespace: stringValue(row.namespace),
+    jobName: stringValue(row.job_name),
+    managerUrl: stringValue(row.manager_url),
+    image: stringValue(row.image),
+    sourceCommit: stringValue(row.source_commit),
+    serviceAccountName: nullableString(row.service_account_name),
+    result: jsonRecord(row.result),
+    createdAt: iso(row.created_at),
+    updatedAt: iso(row.updated_at),
+  };
+}
+
 function metadataForRunner(runner: RunnerRecord): JsonRecord {
   const { id: _id, runId: _runId, attemptId: _attemptId, backendProfile: _backendProfile, placement: _placement, sourceCommit: _sourceCommit, registeredAt: _registeredAt, heartbeatAt: _heartbeatAt, ...metadata } = runner;
   return redactJson(metadata);
 }
 
-function isTerminalStatus(status: RunRecord["status"]): boolean {
-  return status === "completed" || status === "failed" || status === "cancelled";
-}
-
 function stringValue(value: unknown): string {
   return typeof value === "string" ? value : String(value ?? "");
 }

src/mgr/result.ts

@@ -0,0 +1,137 @@
+import type { AgentRunStore } from "./store.js";
+import type { CommandRecord, JsonRecord, JsonValue, RunEvent, RunRecord, RunnerJobRecord, TerminalStatus } from "../common/types.js";
+
+export async function buildRunResult(store: AgentRunStore, runId: string, commandId?: string): Promise<JsonRecord> {
+  const run = await store.getRun(runId);
+  const command = await selectCommand(store, runId, commandId);
+  const events = await store.listEvents(runId, 0, 500);
+  const jobs = await store.listRunnerJobs(runId, command?.id);
+  const latestJob = jobs.at(-1) ?? null;
+  const terminal = terminalFromEvents(events) ?? run.terminalStatus;
+  const failureKind = run.failureKind ?? failureKindFromEvents(events);
+  const reply = assistantReply(events);
+  const blocker = terminal === "blocked" || terminal === "failed" ? { failureKind, message: run.failureMessage ?? messageFromEvents(events) } : null;
+  return {
+    runId: run.id,
+    commandId: command?.id ?? commandId ?? null,
+    attemptId: latestJob?.attemptId ?? attemptFromEvents(events),
+    runnerId: latestJob?.runnerId ?? null,
+    jobName: latestJob?.jobName ?? null,
+    namespace: latestJob?.namespace ?? null,
+    status: command?.state ?? run.status,
+    runStatus: run.status,
+    commandState: command?.state ?? null,
+    terminalStatus: terminal,
+    reply,
+    failureKind,
+    failureMessage: run.failureMessage ?? messageFromEvents(events),
+    blocker,
+    lastSeq: events.at(-1)?.seq ?? 0,
+    eventCount: events.length,
+    artifactSummary: artifactSummary(events),
+    sessionRef: sessionSummary(run),
+    resourceBundleRef: resourceBundleSummary(run, events),
+    runnerJobCount: jobs.length,
+  };
+}
+
+async function selectCommand(store: AgentRunStore, runId: string, commandId?: string): Promise<CommandRecord | null> {
+  if (commandId) {
+    const command = await store.getCommand(commandId);
+    return command.runId === runId ? command : null;
+  }
+  const commands = await store.listCommands(runId, 0, 100);
+  return commands.at(-1) ?? null;
+}
+
+function terminalFromEvents(events: RunEvent[]): TerminalStatus | null {
+  for (const event of [...events].reverse()) {
+    if (event.type !== "terminal_status") continue;
+    const value = event.payload.terminalStatus;
+    if (value === "completed" || value === "failed" || value === "blocked" || value === "cancelled") return value;
+  }
+  return null;
+}
+
+function failureKindFromEvents(events: RunEvent[]): string | null {
+  for (const event of [...events].reverse()) {
+    const value = event.payload.failureKind;
+    if (typeof value === "string") return value;
+  }
+  return null;
+}
+
+function messageFromEvents(events: RunEvent[]): string | null {
+  for (const event of [...events].reverse()) {
+    const value = event.payload.message;
+    if (typeof value === "string" && value.length > 0) return value;
+  }
+  return null;
+}
+
+function assistantReply(events: RunEvent[]): string {
+  return events
+    .filter((event) => event.type === "assistant_message")
+    .map((event) => textPayload(event.payload))
+    .filter((text) => text.length > 0)
+    .join("");
+}
+
+function textPayload(payload: JsonRecord): string {
+  for (const key of ["text", "content", "delta"]) {
+    const value = payload[key];
+    if (typeof value === "string") return value;
+  }
+  return "";
+}
+
+function artifactSummary(events: RunEvent[]): JsonRecord {
+  let commandOutputEvents = 0;
+  let diffEvents = 0;
+  let toolCallEvents = 0;
+  let outputChars = 0;
+  let truncatedEvents = 0;
+  for (const event of events) {
+    if (event.type === "command_output") {
+      commandOutputEvents += 1;
+      const text = textPayload(event.payload);
+      outputChars += text.length;
+      if (event.payload.truncated === true) truncatedEvents += 1;
+    }
+    if (event.type === "diff") diffEvents += 1;
+    if (event.type === "tool_call") toolCallEvents += 1;
+  }
+  return { commandOutputEvents, diffEvents, toolCallEvents, outputChars, truncatedEvents };
+}
+
+function attemptFromEvents(events: RunEvent[]): string | null {
+  for (const event of [...events].reverse()) {
+    const value = event.payload.attemptId;
+    if (typeof value === "string") return value;
+  }
+  return null;
+}
+
+function sessionSummary(run: RunRecord): JsonRecord | null {
+  if (!run.sessionRef) return null;
+  return {
+    sessionId: run.sessionRef.sessionId,
+    conversationId: run.sessionRef.conversationId ?? null,
+    threadId: run.sessionRef.threadId ?? null,
+    expiresAt: run.sessionRef.expiresAt ?? null,
+    metadataKeys: Object.keys(run.sessionRef.metadata ?? {}).sort(),
+    valuesPrinted: false,
+  };
+}
+
+function resourceBundleSummary(run: RunRecord, events: RunEvent[]): JsonRecord | null {
+  if (!run.resourceBundleRef) return null;
+  const materialized = events.find((event) => event.type === "backend_status" && event.payload.phase === "resource-bundle-materialized")?.payload ?? null;
+  return {
+    kind: run.resourceBundleRef.kind,
+    repoUrl: run.resourceBundleRef.repoUrl,
+    commitId: run.resourceBundleRef.commitId,
+    subdir: run.resourceBundleRef.subdir ?? null,
+    materialized: materialized as JsonValue,
+  };
+}

src/mgr/server.ts

@@ -7,6 +7,7 @@ import { AgentRunError, errorToJson } from "../common/errors.js";
 import { asRecord, validateCreateCommand, validateCreateRun } from "../common/validation.js";
 import type { ApiErrorBody, ApiOkBody, JsonRecord, JsonValue, RunEvent } from "../common/types.js";
 import { createKubernetesRunnerJob } from "./kubernetes-runner-job.js";
+import { buildRunResult } from "./result.js";
 
 export interface ManagerServerOptions {
   store?: AgentRunStore;
@@ -75,6 +76,14 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
     const limit = integerQuery(url, "limit", 100);
     return { items: await store.listEvents(eventMatch[1] ?? "", afterSeq, limit) as unknown as JsonValue };
   }
+  const runResultMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/result$/u);
+  if (method === "GET" && runResultMatch) return await buildRunResult(store, runResultMatch[1] ?? "", url.searchParams.get("commandId") ?? undefined) as JsonValue;
+  const runCancelMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/cancel$/u);
+  if (method === "POST" && runCancelMatch) {
+    const record = body === null ? {} : asRecord(body, "cancel");
+    const reason = typeof record.reason === "string" && record.reason.trim().length > 0 ? record.reason.trim() : undefined;
+    return await store.cancelRun(runCancelMatch[1] ?? "", reason) as unknown as JsonValue;
+  }
   const commandCreateMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/commands$/u);
   if (method === "POST" && commandCreateMatch) return await store.createCommand(commandCreateMatch[1] ?? "", validateCreateCommand(body)) as unknown as JsonValue;
   if (method === "GET" && commandCreateMatch) return { items: await store.listCommands(commandCreateMatch[1] ?? "", integerQuery(url, "afterSeq", 0), integerQuery(url, "limit", 20)) as unknown as JsonValue };
@@ -97,6 +106,8 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
   }
   const commandShowMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/commands\/([^/]+)$/u);
   if (method === "GET" && commandShowMatch) return await store.getCommand(commandShowMatch[2] ?? "") as unknown as JsonValue;
+  const commandResultMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/commands\/([^/]+)\/result$/u);
+  if (method === "GET" && commandResultMatch) return await buildRunResult(store, commandResultMatch[1] ?? "", commandResultMatch[2] ?? "") as JsonValue;
   if (method === "POST" && path === "/api/v1/runners/register") return await store.registerRunner(asRecord(body ?? {}, "runner")) as unknown as JsonValue;
   const claimMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/claim$/u);
   if (method === "POST" && claimMatch) {
@@ -122,7 +133,13 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
   if (method === "PATCH" && statusMatch) {
     const record = asRecord(body, "status");
     const terminalStatus = record.terminalStatus === "completed" || record.terminalStatus === "failed" || record.terminalStatus === "blocked" || record.terminalStatus === "cancelled" ? record.terminalStatus : "failed";
-    return await store.finishRun(statusMatch[1] ?? "", { terminalStatus, failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null, failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null }) as unknown as JsonValue;
+    return await store.finishRun(statusMatch[1] ?? "", {
+      terminalStatus,
+      failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null,
+      failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null,
+      ...(typeof record.threadId === "string" ? { threadId: record.threadId } : {}),
+      ...(typeof record.turnId === "string" ? { turnId: record.turnId } : {}),
+    }) as unknown as JsonValue;
   }
   const ackMatch = path.match(/^\/api\/v1\/commands\/([^/]+)\/ack$/u);
   if (method === "POST" && ackMatch) return await store.ackCommand(ackMatch[1] ?? "") as unknown as JsonValue;
@@ -132,6 +149,12 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
     const terminalStatus = record.terminalStatus === "completed" || record.terminalStatus === "failed" || record.terminalStatus === "blocked" || record.terminalStatus === "cancelled" ? record.terminalStatus : "failed";
     return await store.finishCommand(commandStatusMatch[1] ?? "", { terminalStatus, failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null, failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null }) as unknown as JsonValue;
   }
+  const commandCancelMatch = path.match(/^\/api\/v1\/commands\/([^/]+)\/cancel$/u);
+  if (method === "POST" && commandCancelMatch) {
+    const record = body === null ? {} : asRecord(body, "cancel");
+    const reason = typeof record.reason === "string" && record.reason.trim().length > 0 ? record.reason.trim() : undefined;
+    return await store.cancelCommand(commandCancelMatch[1] ?? "", reason) as unknown as JsonValue;
+  }
   throw new AgentRunError("schema-invalid", `unsupported route ${method} ${path}`, { httpStatus: 404 });
 }
 

src/mgr/store.ts

@@ -1,4 +1,4 @@
-import type { BackendProfile, BackendTurnResult, CommandRecord, CreateCommandInput, CreateRunInput, FailureKind, JsonRecord, RunEvent, RunnerRecord, RunRecord, TerminalStatus } from "../common/types.js";
+import type { BackendProfile, BackendTurnResult, CommandRecord, CreateCommandInput, CreateRunInput, FailureKind, JsonRecord, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, SessionRecord, SessionRef, TerminalStatus } from "../common/types.js";
 import { AgentRunError } from "../common/errors.js";
 import { newId, nowIso, stableHash } from "../common/validation.js";
 import { redactJson } from "../common/redaction.js";
@@ -26,16 +26,38 @@ export interface AgentRunStore {
   getCommand(commandId: string): MaybePromise<CommandRecord>;
   listCommands(runId: string, afterSeq: number, limit: number): MaybePromise<CommandRecord[]>;
   registerRunner(input: Partial<RunnerRecord>): MaybePromise<RunnerRecord>;
+  listRunnerJobs(runId: string, commandId?: string): MaybePromise<RunnerJobRecord[]>;
+  getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): MaybePromise<RunnerJobRecord | null>;
+  saveRunnerJob(input: SaveRunnerJobInput): MaybePromise<RunnerJobRecord>;
   claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
   heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
   ackCommand(commandId: string): MaybePromise<CommandRecord>;
   finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): MaybePromise<CommandRecord>;
   appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): MaybePromise<RunEvent>;
-  finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): MaybePromise<RunRecord>;
+  finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): MaybePromise<RunRecord>;
+  cancelRun(runId: string, reason?: string): MaybePromise<RunRecord>;
+  cancelCommand(commandId: string, reason?: string): MaybePromise<CommandRecord>;
+  getSession(sessionId: string): MaybePromise<SessionRecord | null>;
   backends(): MaybePromise<JsonRecord[]>;
   close?(): MaybePromise<void>;
 }
 
+export interface SaveRunnerJobInput {
+  runId: string;
+  commandId: string;
+  idempotencyKey?: string | null;
+  payloadHash: string;
+  attemptId: string;
+  runnerId: string;
+  namespace: string;
+  jobName: string;
+  managerUrl: string;
+  image: string;
+  sourceCommit: string;
+  serviceAccountName?: string | null;
+  result: JsonRecord;
+}
+
 export async function openAgentRunStoreFromEnv(env: NodeJS.ProcessEnv = process.env): Promise<AgentRunStore> {
   const databaseUrl = env.DATABASE_URL?.trim();
   if (databaseUrl) {
@@ -52,6 +74,8 @@ export class MemoryAgentRunStore implements AgentRunStore {
   private readonly commands = new Map<string, CommandRecord>();
   private readonly eventsByRun = new Map<string, RunEvent[]>();
   private readonly runners = new Map<string, RunnerRecord>();
+  private readonly sessions = new Map<string, SessionRecord>();
+  private readonly runnerJobs = new Map<string, RunnerJobRecord>();
 
   health(): StoreHealth {
     return { adapter: "memory-self-test", ready: true, reachable: true, migrationReady: true, migrationId: "memory-self-test", failureKind: null, message: null, credentialValuesPrinted: false };
@@ -59,10 +83,11 @@ export class MemoryAgentRunStore implements AgentRunStore {
 
   createRun(input: CreateRunInput): RunRecord {
     const at = nowIso();
-    const run: RunRecord = { ...input, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null };
+    const sessionRef = this.resolveSessionForRun(input, at);
+    const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null };
     this.runs.set(run.id, run);
     this.eventsByRun.set(run.id, []);
-    this.appendEvent(run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile });
+    this.appendEvent(run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile, sessionRef: summarizeSessionRef(run.sessionRef ?? null), resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null) });
     return run;
   }
 
@@ -113,9 +138,35 @@ export class MemoryAgentRunStore implements AgentRunStore {
     return runner;
   }
 
+  listRunnerJobs(runId: string, commandId?: string): RunnerJobRecord[] {
+    this.getRun(runId);
+    return Array.from(this.runnerJobs.values())
+      .filter((job) => job.runId === runId && (!commandId || job.commandId === commandId))
+      .sort((a, b) => a.createdAt.localeCompare(b.createdAt));
+  }
+
+  getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): RunnerJobRecord | null {
+    const existing = Array.from(this.runnerJobs.values()).find((job) => job.runId === runId && job.idempotencyKey === idempotencyKey);
+    if (!existing) return null;
+    if (existing.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "runner job idempotency key reused with different payload", { httpStatus: 409 });
+    return existing;
+  }
+
+  saveRunnerJob(input: SaveRunnerJobInput): RunnerJobRecord {
+    if (input.idempotencyKey) {
+      const existing = this.getRunnerJobByIdempotencyKey(input.runId, input.idempotencyKey, input.payloadHash);
+      if (existing) return existing;
+    }
+    const at = nowIso();
+    const record: RunnerJobRecord = { ...input, id: newId("rjob"), idempotencyKey: input.idempotencyKey ?? null, serviceAccountName: input.serviceAccountName ?? null, createdAt: at, updatedAt: at };
+    this.runnerJobs.set(record.id, record);
+    return record;
+  }
+
   claimRun(runId: string, runnerId: string, leaseMs: number): RunRecord {
     const run = this.getRun(runId);
-    if (run.claimedBy && run.claimedBy !== runnerId && run.status !== "completed" && run.status !== "failed" && run.status !== "cancelled") throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
+    if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 });
+    if (run.claimedBy && run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
     const next = this.updateRun(runId, { status: "claimed", claimedBy: runnerId, leaseExpiresAt: new Date(Date.now() + leaseMs).toISOString() });
     this.appendEvent(runId, "backend_status", { phase: "run-claimed", runnerId });
     return next;
@@ -123,12 +174,14 @@ export class MemoryAgentRunStore implements AgentRunStore {
 
   heartbeat(runId: string, runnerId: string, leaseMs: number): RunRecord {
     const run = this.getRun(runId);
+    if (isTerminalRunStatus(run.status)) return run;
     if (run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is not claimed by ${runnerId}`, { httpStatus: 409 });
     return this.updateRun(runId, { leaseExpiresAt: new Date(Date.now() + leaseMs).toISOString() });
   }
 
   ackCommand(commandId: string): CommandRecord {
     const command = this.getCommand(commandId);
+    if (isTerminalCommandState(command.state) || command.state === "acknowledged") return command;
     const next = { ...command, state: "acknowledged" as const, acknowledgedAt: nowIso(), updatedAt: nowIso() };
     this.commands.set(commandId, next);
     return next;
@@ -136,6 +189,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
 
   finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): CommandRecord {
     const command = this.getCommand(commandId);
+    if (isTerminalCommandState(command.state)) return command;
     const next = { ...command, state: commandStateFromTerminal(result.terminalStatus), updatedAt: nowIso() };
     this.commands.set(commandId, next);
     this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: next.state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
@@ -151,13 +205,45 @@ export class MemoryAgentRunStore implements AgentRunStore {
     return event;
   }
 
-  finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): RunRecord {
+  finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): RunRecord {
+    const existing = this.getRun(runId);
+    if (isTerminalRunStatus(existing.status)) return existing;
     const status = statusFromTerminal(result.terminalStatus);
     const next = this.updateRun(runId, { status, terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage });
+    if (result.threadId && next.sessionRef?.sessionId) this.upsertSessionThread(next, result.threadId, result.turnId ?? null);
     this.appendEvent(runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage });
     return next;
   }
 
+  cancelRun(runId: string, reason = "cancel requested"): RunRecord {
+    const run = this.getRun(runId);
+    if (isTerminalRunStatus(run.status)) return run;
+    const at = nowIso();
+    for (const command of Array.from(this.commands.values()).filter((item) => item.runId === runId && !isTerminalCommandState(item.state))) {
+      const cancelled = { ...command, state: "cancelled" as const, updatedAt: at };
+      this.commands.set(command.id, cancelled);
+      this.appendEvent(runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
+    }
+    this.appendEvent(runId, "backend_status", { phase: "cancel-requested", reason });
+    const next = this.updateRun(runId, { status: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: reason });
+    this.appendEvent(runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
+    return next;
+  }
+
+  cancelCommand(commandId: string, reason = "cancel requested"): CommandRecord {
+    const command = this.getCommand(commandId);
+    if (isTerminalCommandState(command.state)) return command;
+    const next = { ...command, state: "cancelled" as const, updatedAt: nowIso() };
+    this.commands.set(commandId, next);
+    this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
+    this.cancelRun(command.runId, reason);
+    return next;
+  }
+
+  getSession(sessionId: string): SessionRecord | null {
+    return this.sessions.get(sessionId) ?? null;
+  }
+
   backends(): JsonRecord[] {
     return backendCapabilities();
   }
@@ -168,6 +254,48 @@ export class MemoryAgentRunStore implements AgentRunStore {
     this.runs.set(runId, next);
     return next;
   }
+
+  private resolveSessionForRun(input: CreateRunInput, at: string): SessionRef | null {
+    if (!input.sessionRef) return null;
+    const existing = this.sessions.get(input.sessionRef.sessionId);
+    if (existing) return sessionRefFromRecord(existing, input.sessionRef);
+    const record: SessionRecord = {
+      sessionId: input.sessionRef.sessionId,
+      tenantId: input.tenantId,
+      projectId: input.projectId,
+      backendProfile: input.backendProfile,
+      conversationId: input.sessionRef.conversationId ?? null,
+      threadId: input.sessionRef.threadId ?? null,
+      metadata: input.sessionRef.metadata ?? {},
+      createdAt: at,
+      updatedAt: at,
+      expiresAt: input.sessionRef.expiresAt ?? null,
+    };
+    this.sessions.set(record.sessionId, record);
+    return sessionRefFromRecord(record, input.sessionRef);
+  }
+
+  private upsertSessionThread(run: RunRecord, threadId: string, turnId: string | null): void {
+    if (!run.sessionRef?.sessionId) return;
+    const at = nowIso();
+    const existing = this.sessions.get(run.sessionRef.sessionId);
+    const record: SessionRecord = {
+      sessionId: run.sessionRef.sessionId,
+      tenantId: run.tenantId,
+      projectId: run.projectId,
+      backendProfile: run.backendProfile,
+      conversationId: run.sessionRef.conversationId ?? existing?.conversationId ?? null,
+      threadId,
+      metadata: { ...(existing?.metadata ?? {}), ...(run.sessionRef.metadata ?? {}), ...(turnId ? { lastTurnId: turnId } : {}) },
+      createdAt: existing?.createdAt ?? at,
+      updatedAt: at,
+      expiresAt: run.sessionRef.expiresAt ?? existing?.expiresAt ?? null,
+    };
+    this.sessions.set(record.sessionId, record);
+    const nextSessionRef = sessionRefFromRecord(record, run.sessionRef);
+    this.updateRun(run.id, { sessionRef: nextSessionRef });
+    this.appendEvent(run.id, "backend_status", { phase: "session-updated", sessionRef: summarizeSessionRef(nextSessionRef), turnId });
+  }
 }
 
 export function statusFromTerminal(terminalStatus: TerminalStatus): RunRecord["status"] {
@@ -182,3 +310,47 @@ export function commandStateFromTerminal(terminalStatus: TerminalStatus): Comman
   if (terminalStatus === "cancelled") return "cancelled";
   return "failed";
 }
+
+export function isTerminalRunStatus(status: RunRecord["status"]): boolean {
+  return status === "completed" || status === "failed" || status === "blocked" || status === "cancelled";
+}
+
+export function isTerminalCommandState(state: CommandRecord["state"]): boolean {
+  return state === "completed" || state === "failed" || state === "cancelled";
+}
+
+export function sessionRefFromRecord(record: SessionRecord, fallback: SessionRef): SessionRef {
+  return {
+    sessionId: record.sessionId,
+    ...(record.conversationId ? { conversationId: record.conversationId } : fallback.conversationId ? { conversationId: fallback.conversationId } : {}),
+    ...(record.threadId ? { threadId: record.threadId } : fallback.threadId ? { threadId: fallback.threadId } : {}),
+    ...(record.expiresAt ? { expiresAt: record.expiresAt } : fallback.expiresAt ? { expiresAt: fallback.expiresAt } : {}),
+    ...(Object.keys(record.metadata).length > 0 ? { metadata: record.metadata } : fallback.metadata ? { metadata: fallback.metadata } : {}),
+  };
+}
+
+export function summarizeSessionRef(sessionRef: SessionRef | null): JsonRecord | null {
+  if (!sessionRef) return null;
+  return {
+    sessionId: sessionRef.sessionId,
+    conversationId: sessionRef.conversationId ?? null,
+    threadId: sessionRef.threadId ?? null,
+    expiresAt: sessionRef.expiresAt ?? null,
+    metadataKeys: Object.keys(sessionRef.metadata ?? {}).sort(),
+    valuesPrinted: false,
+  };
+}
+
+export function summarizeResourceBundleRef(resourceBundleRef: RunRecord["resourceBundleRef"] | null | undefined): JsonRecord | null {
+  if (!resourceBundleRef) return null;
+  return {
+    kind: resourceBundleRef.kind,
+    repoUrl: resourceBundleRef.repoUrl,
+    commitId: resourceBundleRef.commitId,
+    subdir: resourceBundleRef.subdir ?? null,
+    sparsePathCount: resourceBundleRef.sparsePaths?.length ?? 0,
+    submodules: resourceBundleRef.submodules ?? false,
+    lfs: resourceBundleRef.lfs ?? false,
+    credentialRef: resourceBundleRef.credentialRef ? { name: resourceBundleRef.credentialRef.name, namespace: resourceBundleRef.credentialRef.namespace ?? null, keys: resourceBundleRef.credentialRef.keys ?? [], valuesPrinted: false } : null,
+  };
+}