feat: close AgentRun commander task plane gaps

src/common/types.ts

@@ -309,6 +309,7 @@ export interface CreateQueueTaskInput extends JsonRecord {
   backendProfile: BackendProfile;
   providerId: string | null;
   workspaceRef: WorkspaceRef | null;
+  sessionRef: SessionRef | null;
   executionPolicy: ExecutionPolicy | null;
   resourceBundleRef: ResourceBundleRef | null;
   payload: JsonRecord;

src/common/validation.ts

@@ -318,6 +318,7 @@ export function validateCreateQueueTask(input: unknown): CreateQueueTaskInput {
     backendProfile: backendProfileValue,
     providerId: optionalString(record.providerId) ?? null,
     workspaceRef: record.workspaceRef === undefined || record.workspaceRef === null ? null : requiredRecord(record, "workspaceRef") as CreateQueueTaskInput["workspaceRef"],
+    sessionRef: validateSessionRef(record.sessionRef),
     executionPolicy: record.executionPolicy === undefined || record.executionPolicy === null ? null : validateExecutionPolicy(requiredRecord(record, "executionPolicy")),
     resourceBundleRef: validateResourceBundleRef(record.resourceBundleRef),
     payload: record.payload === undefined ? {} : asRecord(record.payload, "payload"),

src/mgr/postgres-store.ts

@@ -6,7 +6,7 @@ import { redactJson } from "../common/redaction.js";
 import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
 import { newId, nowIso, stableHash } from "../common/validation.js";
 import type { AgentRunStore, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
-import { assertSessionBoundary, buildQueueStats, buildSessionSummary, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, parseSessionCursor, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
+import { assertSessionBoundary, buildQueueStats, buildSessionSummary, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, parseSessionCursor, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
 import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
 import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
 
@@ -329,6 +329,11 @@ CREATE TABLE IF NOT EXISTS agentrun_queue_read_cursors (
 CREATE SEQUENCE IF NOT EXISTS agentrun_queue_version_seq;
 `;
 
+const queueSessionRefMigrationSql = `
+ALTER TABLE agentrun_queue_tasks
+  ADD COLUMN IF NOT EXISTS session_ref jsonb;
+`;
+
 const postgresMigrations: MigrationDefinition[] = [
   {
     id: "001_v01_initial_durable_store",
@@ -375,6 +380,11 @@ const postgresMigrations: MigrationDefinition[] = [
     checksum: checksumSql(dsflashGoModelCatalogBackendProfileMigrationSql),
     sql: dsflashGoModelCatalogBackendProfileMigrationSql,
   },
+  {
+    id: "010_v01_queue_session_ref",
+    checksum: checksumSql(queueSessionRefMigrationSql),
+    sql: queueSessionRefMigrationSql,
+  },
 ];
 
 export function postgresMigrationContract(): JsonRecord {
@@ -905,7 +915,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
   }
 
   async createQueueTask(input: CreateQueueTaskInput): Promise<QueueTaskRecord> {
-    const payloadHash = stableHash(input.payload);
+    const payloadHash = queueTaskPayloadHash(input);
     return this.withTransaction(async (client) => {
       if (input.idempotencyKey) {
         const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE tenant_id = $1 AND project_id = $2 AND idempotency_key = $3 FOR UPDATE", [input.tenantId, input.projectId, input.idempotencyKey]);
@@ -917,12 +927,14 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       }
       const at = nowIso();
       const version = await this.nextQueueVersion(client);
-      const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version, payloadHash, latestAttempt: null, sessionPath: null, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
+      const sessionId = input.sessionRef?.sessionId ?? null;
+      const sessionPath = sessionId ? `/api/v1/sessions/${encodeURIComponent(sessionId)}` : null;
+      const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version, payloadHash, latestAttempt: null, sessionPath, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
       const inserted = await client.query(
-        `INSERT INTO agentrun_queue_tasks (id, tenant_id, project_id, queue, lane, title, priority, state, version, backend_profile, provider_id, workspace_ref, execution_policy, resource_bundle_ref, payload, payload_hash, references_json, metadata, idempotency_key, latest_attempt, session_path, created_at, updated_at, cancelled_at, cancel_reason)
-         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, $13::jsonb, $14::jsonb, $15::jsonb, $16, $17::jsonb, $18::jsonb, $19, $20::jsonb, $21, $22, $23, $24, $25)
+        `INSERT INTO agentrun_queue_tasks (id, tenant_id, project_id, queue, lane, title, priority, state, version, backend_profile, provider_id, workspace_ref, session_ref, execution_policy, resource_bundle_ref, payload, payload_hash, references_json, metadata, idempotency_key, latest_attempt, session_path, created_at, updated_at, cancelled_at, cancel_reason)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, $13::jsonb, $14::jsonb, $15::jsonb, $16::jsonb, $17, $18::jsonb, $19::jsonb, $20, $21::jsonb, $22, $23, $24, $25, $26)
          RETURNING *`,
-        [task.id, task.tenantId, task.projectId, task.queue, task.lane, task.title, task.priority, task.state, task.version, task.backendProfile, task.providerId, JSON.stringify(task.workspaceRef), JSON.stringify(task.executionPolicy), JSON.stringify(task.resourceBundleRef), JSON.stringify(task.payload), task.payloadHash, JSON.stringify(task.references), JSON.stringify(task.metadata), task.idempotencyKey ?? null, JSON.stringify(task.latestAttempt), task.sessionPath, task.createdAt, task.updatedAt, task.cancelledAt, task.cancelReason],
+        [task.id, task.tenantId, task.projectId, task.queue, task.lane, task.title, task.priority, task.state, task.version, task.backendProfile, task.providerId, JSON.stringify(task.workspaceRef), JSON.stringify(task.sessionRef), JSON.stringify(task.executionPolicy), JSON.stringify(task.resourceBundleRef), JSON.stringify(task.payload), task.payloadHash, JSON.stringify(task.references), JSON.stringify(task.metadata), task.idempotencyKey ?? null, JSON.stringify(task.latestAttempt), task.sessionPath, task.createdAt, task.updatedAt, task.cancelledAt, task.cancelReason],
       );
       return queueTaskFromRow(inserted.rows[0]);
     });
@@ -1404,6 +1416,7 @@ function queueTaskFromRow(row: QueryResultRow): QueueTaskRecord {
     backendProfile: stringValue(row.backend_profile) as BackendProfile,
     providerId: nullableString(row.provider_id),
     workspaceRef: jsonValue(row.workspace_ref) as QueueTaskRecord["workspaceRef"],
+    sessionRef: jsonValue(row.session_ref) as QueueTaskRecord["sessionRef"],
     executionPolicy: jsonValue(row.execution_policy) as QueueTaskRecord["executionPolicy"],
     resourceBundleRef: jsonValue(row.resource_bundle_ref) as QueueTaskRecord["resourceBundleRef"],
     payload: jsonRecord(row.payload),

src/mgr/queue-dispatch.ts

@@ -99,6 +99,7 @@ function buildRunInput(task: QueueTaskRecord, input: JsonRecord): CreateRunInput
     tenantId: task.tenantId,
     projectId: task.projectId,
     workspaceRef: task.workspaceRef,
+    sessionRef: task.sessionRef,
     resourceBundleRef: task.resourceBundleRef,
     providerId: task.providerId,
     backendProfile: task.backendProfile,

src/mgr/store.ts

@@ -451,7 +451,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
   }
 
   createQueueTask(input: CreateQueueTaskInput): QueueTaskRecord {
-    const payloadHash = stableHash(input.payload);
+    const payloadHash = queueTaskPayloadHash(input);
     if (input.idempotencyKey) {
       const existing = Array.from(this.queueTasks.values()).find((task) => task.tenantId === input.tenantId && task.projectId === input.projectId && task.idempotencyKey === input.idempotencyKey);
       if (existing) {
@@ -460,7 +460,9 @@ export class MemoryAgentRunStore implements AgentRunStore {
       }
     }
     const at = nowIso();
-    const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version: this.nextQueueVersion(), payloadHash, latestAttempt: null, sessionPath: null, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
+    const sessionId = input.sessionRef?.sessionId ?? null;
+    const sessionPath = sessionId ? `/api/v1/sessions/${encodeURIComponent(sessionId)}` : null;
+    const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version: this.nextQueueVersion(), payloadHash, latestAttempt: null, sessionPath, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
     this.queueTasks.set(task.id, task);
     return task;
   }
@@ -750,6 +752,26 @@ export function queueTaskSort(a: QueueTaskRecord, b: QueueTaskRecord): number {
   return a.createdAt.localeCompare(b.createdAt) || a.id.localeCompare(b.id);
 }
 
+export function queueTaskPayloadHash(input: CreateQueueTaskInput): string {
+  return stableHash({
+    tenantId: input.tenantId,
+    projectId: input.projectId,
+    queue: input.queue,
+    lane: input.lane,
+    title: input.title,
+    priority: input.priority,
+    backendProfile: input.backendProfile,
+    providerId: input.providerId,
+    workspaceRef: input.workspaceRef,
+    sessionRef: input.sessionRef,
+    executionPolicy: input.executionPolicy,
+    resourceBundleRef: input.resourceBundleRef,
+    payload: input.payload,
+    references: input.references,
+    metadata: input.metadata,
+  });
+}
+
 export function buildQueueStats(tasks: QueueTaskRecord[], queue: string | null, generatedAt = nowIso()): QueueStats {
   const byState: Record<string, number> = {};
   const byLane: Record<string, number> = {};

src/selftest/cases/00-redaction-postgres.ts

@@ -13,11 +13,13 @@ const selfTest: SelfTestCase = async () => {
     (error) => error instanceof AgentRunError && error.failureKind === "infra-failed" && error.message.includes("DATABASE_URL is required"),
   );
   const postgresContract = postgresMigrationContract();
-  assert.equal(postgresContract.latestMigrationId, "009_v01_dsflash_go_model_catalog");
+  assert.equal(postgresContract.latestMigrationId, "010_v01_queue_session_ref");
   assert.equal((postgresContract.migrationIds as string[]).includes("008_v01_dsflash_go_backend_profile"), true);
   assert.equal((postgresContract.migrationIds as string[]).includes("009_v01_dsflash_go_model_catalog"), true);
+  assert.equal((postgresContract.migrationIds as string[]).includes("010_v01_queue_session_ref"), true);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"].length > 0);
   assert.ok(typeof (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"].length > 0);
+  assert.ok(typeof (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"].length > 0);
   assert.equal((postgresContract.checksums as Record<string, string>)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c");
   assert.ok(Array.isArray(postgresContract.requiredTables));
   assert.ok(postgresContract.requiredTables.includes("agentrun_schema_migrations"));

src/selftest/cases/70-queue-q1.ts

@@ -19,6 +19,7 @@ const selfTest: SelfTestCase = async (context) => {
       backendProfile: "codex",
       providerId: "G14",
       workspaceRef: { kind: "host-path", path: context.workspace },
+      sessionRef: { sessionId: "sess_queue_q1_selftest", metadata: { source: "queue-q1-self-test" } },
       executionPolicy: null,
       resourceBundleRef: null,
       payload: { prompt: "hello" },
@@ -28,10 +29,15 @@ const selfTest: SelfTestCase = async (context) => {
     };
     const created = await client.post("/api/v1/queue/tasks", input) as QueueTaskRecord;
     assert.equal(created.state, "pending");
-    assert.equal(created.sessionPath, null);
+    assert.equal(created.sessionRef?.sessionId, "sess_queue_q1_selftest");
+    assert.equal(created.sessionPath, "/api/v1/sessions/sess_queue_q1_selftest");
     assert.equal(created.latestAttempt, null);
     const duplicate = await client.post("/api/v1/queue/tasks", input) as QueueTaskRecord;
     assert.equal(duplicate.id, created.id);
+    await assert.rejects(
+      () => client.post("/api/v1/queue/tasks", { ...input, sessionRef: { sessionId: "sess_queue_q1_other" } }),
+      (error) => error instanceof Error && error.message.includes("idempotency key reused"),
+    );
 
     const listed = await client.get("/api/v1/queue/tasks?queue=dev&limit=10") as QueueTaskListResult;
     assert.equal(listed.count, 1);
@@ -39,7 +45,7 @@ const selfTest: SelfTestCase = async (context) => {
 
     const shown = await client.get(`/api/v1/queue/tasks/${created.id}`) as QueueTaskRecord;
     assert.equal(shown.title, "Q1 queue task");
-    assert.equal(shown.sessionPath, null);
+    assert.equal(shown.sessionPath, "/api/v1/sessions/sess_queue_q1_selftest");
 
     const stats = await client.get("/api/v1/queue/stats?queue=dev") as QueueStats;
     assert.equal(stats.total, 1);

src/selftest/cases/75-queue-q2-dispatch.ts

@@ -44,6 +44,7 @@ console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kin
       backendProfile: "codex",
       providerId: "G14",
       workspaceRef: { kind: "host-path", path: context.workspace },
+      sessionRef: { sessionId: "sess_queue_q2_dispatch_selftest", metadata: { source: "queue-q2-self-test" } },
       executionPolicy: {
         sandbox: "workspace-write",
         approval: "never",
@@ -64,9 +65,12 @@ console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kin
     assert.equal(dispatched.latestAttempt.runId, dispatched.run.id);
     assert.equal(dispatched.latestAttempt.commandId, dispatched.command.id);
     assert.ok(dispatched.latestAttempt.runnerJobId);
+    assert.equal(dispatched.latestAttempt.sessionId, "sess_queue_q2_dispatch_selftest");
+    assert.equal(dispatched.latestAttempt.sessionPath, "/api/v1/sessions/sess_queue_q2_dispatch_selftest");
     assert.equal(dispatched.task.state, "running");
     assert.equal(dispatched.task.latestAttempt?.attemptId, "attempt_queue_q2_selftest");
-    assert.equal(dispatched.task.sessionPath, null);
+    assert.equal(dispatched.task.sessionPath, "/api/v1/sessions/sess_queue_q2_dispatch_selftest");
+    assert.equal(dispatched.run.sessionRef?.sessionId, "sess_queue_q2_dispatch_selftest");
 
     const shown = await client.get(`/api/v1/queue/tasks/${created.id}`) as QueueTaskRecord;
     assert.equal(shown.state, "running");
@@ -88,6 +92,7 @@ console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kin
     assert.equal(refreshed.state, "completed");
     assert.equal(refreshed.latestAttempt?.state, "completed");
     assert.equal(refreshed.latestAttempt?.runId, dispatched.run.id);
+    assert.equal(refreshed.latestAttempt?.sessionPath, "/api/v1/sessions/sess_queue_q2_dispatch_selftest");
     const manifest = JSON.parse(await readFile(createdManifest, "utf8")) as JsonRecord;
     assert.ok(JSON.stringify(manifest).includes(dispatched.run.id));
     assertNoSecretLeak(dispatched);

src/selftest/cases/90-runner-image-tools.ts

@@ -1,19 +1,35 @@
 import assert from "node:assert/strict";
 import { readFile } from "node:fs/promises";
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
 import path from "node:path";
 import type { SelfTestCase } from "../harness.js";
 
 const requiredRunnerPackages = Object.freeze(["git", "openssh-client", "ripgrep"]);
+const execFileAsync = promisify(execFile);
 
 const selfTest: SelfTestCase = async (context) => {
   const containerfile = await readFile(path.join(context.root, "deploy/container/Containerfile"), "utf8");
   const apkPackages = installedApkPackages(containerfile);
+  const tran = await readFile(path.join(context.root, "tools/tran"), "utf8");
+  const trans = await readFile(path.join(context.root, "tools/trans"), "utf8");
 
   for (const packageName of requiredRunnerPackages) {
     assert.equal(apkPackages.has(packageName), true, `runner image must install ${packageName}`);
   }
 
-  return { name: "90-runner-image-tools", tests: ["runner image installs required CLI tools"] };
+  assert.equal(tran.startsWith("#!/usr/bin/env bun\n"), true, "tools/tran must be a shebang executable discovered by gitbundle tools");
+  assert.equal(trans.startsWith("#!/bin/sh\n"), true, "tools/trans must be a shebang executable discovered by gitbundle tools");
+  assert.equal(tran.includes("UNIDESK_SSH_CLIENT_TOKEN"), true, "tools/tran must require the scoped UniDesk SSH client token");
+  assert.equal(tran.includes("/ws/ssh"), true, "tools/tran must use the frontend SSH WebSocket path");
+
+  const help = await execFileAsync(path.join(context.root, "tools/tran"), ["--help"], { cwd: context.root, timeout: 10_000 });
+  const parsed = JSON.parse(help.stdout) as { ok?: boolean; supported?: string[]; valuesPrinted?: boolean };
+  assert.equal(parsed.ok, true);
+  assert.equal(parsed.valuesPrinted, false);
+  assert.equal(parsed.supported?.some((line) => line.includes("script")), true);
+
+  return { name: "90-runner-image-tools", tests: ["runner image installs required CLI tools", "gitbundle tran tools are executable and documented"] };
 };
 
 function installedApkPackages(containerfile: string): Set<string> {