sync: update from private repo (ea88916)

src/bridge/console-session-api.test.ts

@@ -0,0 +1,313 @@
+/**
+ * Tests for POST /api/local/tasks/:taskId/console/session — the
+ * user-initiated SSH console session-open endpoint (Task 2).
+ *
+ * Strategy: mirror the hermetic stub-subsystem approach from
+ * ssh-console.test.ts. We do NOT dial real SSH. The shell-open collaborator
+ * (`openShellChannel`) is faked to return a channel/client/fingerprint, and a
+ * real `SessionRegistry` is used so we can assert `registry.get(taskId)`
+ * actually returns a session after a successful open. The access gate
+ * (preflight + accessResolver + host-key check) runs for real inside the
+ * shared `openConsoleSession` core the endpoint calls — the test only fakes
+ * the network dial, not the gate.
+ */
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+import express from 'express';
+import request from 'supertest';
+import { createConsoleSessionRouter } from './console-ws-api.js';
+import { preflight, type SshSubsystem } from '../engine/tools/ssh.js';
+import { SessionRegistry } from '../ssh/console-registry.js';
+import type { SimpleTask, SimpleUser } from './console-ws-api.js';
+
+function mkConn(overrides: Partial<{ hostKeyVerifiedAt: string | null; enabled: boolean }> = {}) {
+  return {
+    id: 'conn-1',
+    ownerId: 'owner-1',
+    label: 'test',
+    host: 'localhost',
+    port: 22,
+    username: 'me',
+    privateKeyEnc: Buffer.alloc(0),
+    passphraseEnc: null,
+    keyVersion: 1,
+    keyFingerprint: 'fp-key',
+    hostKeyType: 'ssh-ed25519',
+    hostKeyB64: 'aaa',
+    hostKeyFingerprint: 'fp',
+    hostKeyRecordedAt: '2026-01-01',
+    hostKeyVerifiedAt: overrides.hostKeyVerifiedAt === undefined ? '2026-01-01' : overrides.hostKeyVerifiedAt,
+    hostKeyPending: false,
+    hostKeyPendingB64: null,
+    hostKeyPendingFingerprint: null,
+    hostKeyPendingToken: null,
+    hostKeyPendingSource: null,
+    commandDenyPatterns: null,
+    commandAllowPatterns: null,
+    remotePathPrefix: '/',
+    allowRemoteUnrestricted: true,
+    allowPrivateAddresses: true,
+    enabled: overrides.enabled === undefined ? true : overrides.enabled,
+    disabledByAdmin: false,
+    disabledByAdminReason: null,
+    disabledByAdminAt: null,
+    disabledByAdminUserId: null,
+    createdAt: '2026-01-01',
+    updatedAt: '2026-01-01',
+  };
+}
+
+interface Harness {
+  sub: SshSubsystem;
+  registry: SessionRegistry;
+  openShellChannel: ReturnType<typeof vi.fn>;
+  closeForTaskSpy: ReturnType<typeof vi.fn>;
+}
+
+function mkSub(opts: {
+  conn?: ReturnType<typeof mkConn>;
+  accessAllowed?: boolean;
+  accessReason?: string;
+  isAdmin?: boolean;
+} = {}): Harness {
+  const conn = opts.conn ?? mkConn();
+  const audit = {
+    beginAndComplete: vi.fn().mockReturnValue(1),
+    begin: vi.fn().mockReturnValue(1),
+    complete: vi.fn(),
+    listAuditRows: vi.fn(),
+    pruneOlderThan: vi.fn(),
+    promotePendingToAborted: vi.fn(),
+  };
+
+  const registry = new SessionRegistry({
+    idleTimeoutMs: 60_000,
+    maxSessionDurationMs: 600_000,
+    maxSessionsPerConnection: 3,
+  });
+  // Wrap closeForTask so we can assert force-replace behavior without
+  // depending on a real channel teardown.
+  const closeForTaskSpy = vi.fn(registry.closeForTask.bind(registry));
+  (registry as any).closeForTask = closeForTaskSpy;
+
+  const connectionRepo = {
+    resolveConnection: vi.fn().mockReturnValue(conn),
+  };
+  const abuseRepo = {
+    isLocked: vi.fn().mockReturnValue({ locked: false }),
+    checkAndRecordFailure: vi.fn(),
+    recordSuccess: vi.fn(),
+  };
+  const accessResolver = {
+    resolveAccess: vi.fn().mockReturnValue(
+      opts.accessAllowed === false
+        ? { allowed: false, reason: opts.accessReason ?? 'no_grant' }
+        : { allowed: true },
+    ),
+  };
+  const channel = { write: vi.fn(), end: vi.fn(), setWindow: vi.fn(), on: vi.fn() };
+  const client = { end: vi.fn(), on: vi.fn() };
+  const openShellChannel = vi.fn().mockResolvedValue({
+    channel,
+    client,
+    hostFingerprint: 'SHA256:fake',
+  });
+
+  const sub = {
+    connectionRepo,
+    auditRepo: audit,
+    abuseRepo,
+    accessResolver,
+    sessionRegistry: registry,
+    openShellChannel,
+    getUserAccess: () => ({ isAdmin: opts.isAdmin ?? false, orgIds: [] }),
+    decryptKeyMaterial: () => Buffer.alloc(0),
+    decryptPassphrase: () => null,
+    sshExec: vi.fn(),
+    sshUpload: vi.fn(),
+    sshDownload: vi.fn(),
+    maintenance: { isActive: () => false, snapshot: () => ({ active: false }), enter: () => {}, exit: () => {} } as SshSubsystem['maintenance'],
+    config: {
+      enabled: true,
+      allowPrivateAddresses: true,
+      callTimeoutSeconds: 30,
+      maxOutputBytes: 1024,
+      maxUploadSizeMb: 10,
+      maxDownloadSizeMb: 10,
+      auditRetentionDays: 90,
+      adminBypassesGrants: true,
+      abuseWindowMinutes: 10,
+      abuseFailureThreshold: 5,
+      abuseLockMinutes: 30,
+      console: {
+        enabled: true,
+        idleTimeoutSeconds: 60,
+        maxSessionDurationSeconds: 600,
+        scrollbackBytes: 4096,
+        maxSessionsPerConnection: 3,
+        maxInputBytesPerSend: 1024,
+        autoInjectScreenLines: 24,
+        defaultCols: 80,
+        defaultRows: 24,
+      },
+    },
+  } as unknown as SshSubsystem;
+
+  return { sub, registry, openShellChannel, closeForTaskSpy };
+}
+
+const OWNER: SimpleUser = { id: 'owner-1', role: 'user' };
+
+function mkTask(overrides: Partial<SimpleTask> = {}): SimpleTask {
+  return {
+    id: '1',
+    ownerId: 'owner-1',
+    visibility: 'private',
+    pieceName: 'ssh-console',
+    ...overrides,
+  };
+}
+
+function buildApp(opts: {
+  sub: SshSubsystem;
+  user?: SimpleUser | null;
+  resolveTask?: (id: string, user: SimpleUser) => Promise<SimpleTask | null>;
+}) {
+  const app = express();
+  const user = opts.user === undefined ? OWNER : opts.user;
+  if (user) {
+    app.use((req, _res, next) => { (req as any).user = user; next(); });
+  }
+  app.use(
+    '/api',
+    express.json(),
+    createConsoleSessionRouter({
+      sub: opts.sub,
+      preflight,
+      requireAuth: (_req: any, _res: any, next: any) => next(),
+      resolveTask: opts.resolveTask ?? (async () => mkTask()),
+    }),
+  );
+  return app;
+}
+
+describe('POST /api/local/tasks/:taskId/console/session', () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it('connection owner → 200 and a session is registered for the task', async () => {
+    const h = mkSub();
+    const app = buildApp({ sub: h.sub });
+    const res = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(res.status).toBe(200);
+    expect(res.body.ok).toBe(true);
+    expect(res.body.connection_id).toBe('conn-1');
+    expect(h.openShellChannel).toHaveBeenCalledTimes(1);
+    // The real registry now holds a live session keyed by the task id.
+    expect(h.registry.get('1')).toBeTruthy();
+    expect(h.registry.get('1')!.connectionId).toBe('conn-1');
+  });
+
+  it('non-owner without grant → 403 no_grant', async () => {
+    const h = mkSub({ accessAllowed: false, accessReason: 'no_grant' });
+    const app = buildApp({
+      sub: h.sub,
+      user: { id: 'someone-else', role: 'user' },
+    });
+    const res = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(res.status).toBe(403);
+    expect(res.body.error).toBe('no_grant');
+    expect(h.openShellChannel).not.toHaveBeenCalled();
+    expect(h.registry.get('1')).toBeNull();
+  });
+
+  it('host key not verified → 409 host_key_not_verified', async () => {
+    const h = mkSub({ conn: mkConn({ hostKeyVerifiedAt: null }) });
+    const app = buildApp({ sub: h.sub });
+    const res = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(res.status).toBe(409);
+    expect(res.body.error).toBe('host_key_not_verified');
+    expect(h.openShellChannel).not.toHaveBeenCalled();
+  });
+
+  it('task not visible to the user → 404 task_not_found', async () => {
+    const h = mkSub();
+    const app = buildApp({ sub: h.sub, resolveTask: async () => null });
+    const res = await request(app)
+      .post('/api/local/tasks/999/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(res.status).toBe(404);
+    expect(res.body.error).toBe('task_not_found');
+    expect(h.openShellChannel).not.toHaveBeenCalled();
+  });
+
+  it('already-active same connection → 200 already_active:true (no re-dial)', async () => {
+    const h = mkSub();
+    const app = buildApp({ sub: h.sub });
+    // First open establishes the session.
+    const first = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(first.status).toBe(200);
+    expect(h.openShellChannel).toHaveBeenCalledTimes(1);
+    // Second open on the same connection reuses it.
+    const second = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(second.status).toBe(200);
+    expect(second.body.ok).toBe(true);
+    expect(second.body.already_active).toBe(true);
+    // Still only one dial total.
+    expect(h.openShellChannel).toHaveBeenCalledTimes(1);
+  });
+
+  it('already-active different connection without force_replace → 409, with it → 200', async () => {
+    const h = mkSub();
+    // conn-1 resolves first; for the swap, return conn-2.
+    const conn2 = mkConn();
+    (conn2 as any).id = 'conn-2';
+    const app = buildApp({ sub: h.sub });
+
+    // Open conn-1.
+    const first = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-1' });
+    expect(first.status).toBe(200);
+    expect(h.openShellChannel).toHaveBeenCalledTimes(1);
+
+    // conn-2 now resolves for the swap attempts.
+    (h.sub.connectionRepo.resolveConnection as any).mockReturnValue(conn2);
+
+    // Swap without force → 409 connection_change_requires_force.
+    const noForce = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-2' });
+    expect(noForce.status).toBe(409);
+    expect(noForce.body.error).toBe('connection_change_requires_force');
+    expect(h.openShellChannel).toHaveBeenCalledTimes(1); // no new dial
+
+    // Swap with force → 200, old closed, new dialed.
+    const force = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({ connection_id: 'conn-2', force_replace: true });
+    expect(force.status).toBe(200);
+    expect(force.body.ok).toBe(true);
+    expect(h.closeForTaskSpy).toHaveBeenCalledWith('1', 'connection_change');
+    expect(h.openShellChannel).toHaveBeenCalledTimes(2);
+    expect(h.registry.get('1')!.connectionId).toBe('conn-2');
+  });
+
+  it('missing connection_id → 400', async () => {
+    const h = mkSub();
+    const app = buildApp({ sub: h.sub });
+    const res = await request(app)
+      .post('/api/local/tasks/1/console/session')
+      .send({});
+    expect(res.status).toBe(400);
+    expect(res.body.error).toBe('missing_connection_id');
+  });
+});

src/bridge/console-ws-api.ts

@@ -7,6 +7,8 @@ import type { SessionRegistry } from '../ssh/console-registry.js';
 import type { ConsoleSession } from '../ssh/console-session.js';
 import type { AttachMessage, ServerTextMessage } from '../ssh/console-protocol.js';
 import { checkConsoleInput } from '../ssh/console-deny-check.js';
+import type { OpenConsoleDeps, OpenConsoleResult } from '../engine/tools/ssh-console.js';
+import { openConsoleSession } from '../engine/tools/ssh-console.js';
 
 export interface SimpleUser { id: string; role: 'admin' | 'user' | string }
 export interface SimpleTask { id: string; ownerId: string; visibility: string; pieceName: string }
@@ -289,3 +291,148 @@ export function createConsoleStatusRouter(deps: {
   );
   return r;
 }
+
+/**
+ * Map an `OpenConsoleResult.error` code to an HTTP status. The endpoint
+ * surfaces the same structured `error` code string in the JSON body so the
+ * UI can map it to a localized message. Unknown codes default to 400 (caller
+ * error) except the internal failure codes which default to 500.
+ */
+function statusForOpenError(error: string | undefined): number {
+  switch (error) {
+    case 'connection_not_found':
+      return 404;
+    case 'no_grant':
+      return 403;
+    case 'host_key_not_verified':
+    case 'host_key_mismatch':
+    case 'connection_change_requires_force':
+    case 'abuse_locked':
+    case 'connection_disabled':
+      return 409;
+    case 'decrypt_failed':
+    case 'open_shell_failed':
+      return 500;
+    // missing_connection_id / missing_task_context / no_user_context /
+    // piece_not_configured / piece_not_allowed / preflight_denied and any
+    // other unexpected code → 400 (the request was malformed or denied at a
+    // layer that maps cleanly onto a bad-request response).
+    default:
+      return 400;
+  }
+}
+
+/**
+ * The preflight helper (`preflight` from engine/tools/ssh.ts) returns a flat
+ * `preflight_denied` for several distinct denials (no grant, abuse lock,
+ * disabled connection, connection not found). The REST contract wants the
+ * specific codes, so the handler re-derives them from the human-readable
+ * message that `openConsoleSession` mirrors into `result.message`. This is a
+ * best-effort refinement layered ON TOP of the real gate — the gate itself
+ * (inside openConsoleSession/preflight) is authoritative and unchanged; we
+ * never widen access here, only narrow a generic 400 into a more specific
+ * 4xx for the UI.
+ */
+function refineErrorCode(result: OpenConsoleResult): string {
+  const code = result.error ?? 'unknown';
+  if (code !== 'preflight_denied') return code;
+  // Match the exact human-readable strings `preflight` (engine/tools/ssh.ts)
+  // produces for each denial reason. Order matters: 'disabled' and 'locked'
+  // are checked before the generic access-denied so a disabled/locked
+  // connection isn't mislabeled as no_grant.
+  const msg = (result.message ?? '').toLowerCase();
+  if (msg.includes('does not exist')) return 'connection_not_found';
+  if (msg.includes('is disabled')) return 'connection_disabled';
+  if (msg.includes('temporarily locked')) return 'abuse_locked';
+  if (msg.includes('access denied')) return 'no_grant';
+  return 'preflight_denied';
+}
+
+/**
+ * REST router exposing POST /local/tasks/:taskId/console/session.
+ *
+ * Lets a user open an SSH console PTY session themselves from a task's
+ * Console tab. The session is keyed by localTaskId, so the WS/xterm and the
+ * AI console tools share it automatically. The handler runs the SAME gate as
+ * the agent-facing SshConsoleEnsure tool: it calls the shared
+ * `openConsoleSession` core, which runs the full preflight (piece membership
+ * / access decision / enabled / abuse / host-key) against the task's piece
+ * name. `allowedConnections: ['*']` is passed because the per-piece
+ * allowed-list is an agent-prompt concept; the authoritative gate is the
+ * access resolver against `task.pieceName`, which still runs inside the core.
+ */
+export function createConsoleSessionRouter(deps: {
+  sub: OpenConsoleDeps['sub'];
+  preflight: OpenConsoleDeps['preflight'];
+  requireAuth: any;
+  resolveTask: (taskId: string, user: SimpleUser) => Promise<SimpleTask | null>;
+}): Router {
+  const r = Router();
+  r.post(
+    '/local/tasks/:taskId/console/session',
+    deps.requireAuth,
+    async (req: Request, res: Response) => {
+      const taskId = req.params.taskId!;
+      const user = (req.user as SimpleUser | undefined) ?? null;
+      if (!user) {
+        res.status(401).json({ error: 'unauthenticated' });
+        return;
+      }
+
+      const task = await deps.resolveTask(taskId, user);
+      if (!task) {
+        // Missing OR not visible to this user → opaque 404 (don't leak
+        // task existence across the visibility boundary).
+        res.status(404).json({ error: 'task_not_found' });
+        return;
+      }
+
+      const body = (req.body ?? {}) as {
+        connection_id?: unknown;
+        cols?: unknown;
+        rows?: unknown;
+        force_replace?: unknown;
+      };
+      const connectionId = typeof body.connection_id === 'string' ? body.connection_id : '';
+      if (!connectionId) {
+        res.status(400).json({ error: 'missing_connection_id' });
+        return;
+      }
+      const cols = typeof body.cols === 'number' ? body.cols : undefined;
+      const rows = typeof body.rows === 'number' ? body.rows : undefined;
+
+      const result = await openConsoleSession(
+        { sub: deps.sub, preflight: deps.preflight },
+        {
+          taskId,
+          connectionId,
+          ownerId: task.ownerId || 'local',
+          userId: user.id,
+          pieceName: task.pieceName,
+          // Per-piece allowed-list is an agent-prompt concept; the real gate
+          // is the access resolver against task.pieceName inside the core.
+          allowedConnections: ['*'],
+          cols,
+          rows,
+          forceReplace: body.force_replace === true,
+          initiator: 'user',
+        },
+      );
+
+      if (result.ok) {
+        res.status(200).json({
+          ok: true,
+          connection_id: result.connectionId,
+          cols: result.cols,
+          rows: result.rows,
+          ...(result.alreadyActive ? { already_active: true } : {}),
+        });
+        return;
+      }
+
+      const code = refineErrorCode(result);
+      res.status(statusForOpenError(code)).json({ error: code });
+    },
+  );
+  return r;
+}

src/bridge/server.ts

@@ -88,11 +88,12 @@ import {
 } from '../ssh/session.js';
 import { SessionRegistry } from '../ssh/console-registry.js';
 import { createSshUserRouter, createSshAdminRouter, type SshApiDeps } from './ssh-api.js';
-import { setSshSubsystem } from '../engine/tools/ssh.js';
+import { setSshSubsystem, preflight as sshPreflight, type SshSubsystem } from '../engine/tools/ssh.js';
 import { __setActiveSessionLookup } from '../engine/agent-loop.js';
 import {
   attachConsoleWs,
   createConsoleStatusRouter,
+  createConsoleSessionRouter,
   type SimpleTask,
   type SimpleUser,
 } from './console-ws-api.js';
@@ -629,7 +630,11 @@ export function createCoreServer(opts: CoreServerOptions): {
           // SshDownload tools can access the same repos / session primitives /
           // crypto wrappers that the HTTP layer uses. sessionRegistry is
           // constructed above (hoisted so sshDeps.onAccessRevoked can use it).
-          setSshSubsystem({
+          // Captured in a local const (not just passed to setSshSubsystem)
+          // so the user-initiated console-session REST endpoint can call the
+          // shared openConsoleSession core with the EXACT same `sub` the
+          // agent-facing console tools use — no second SshSubsystem.
+          const sshSubsystem: SshSubsystem = {
             connectionRepo,
             auditRepo,
             abuseRepo,
@@ -651,7 +656,8 @@ export function createCoreServer(opts: CoreServerOptions): {
             config: sshConfig,
             sessionRegistry,
             openShellChannel,
-          });
+          };
+          setSshSubsystem(sshSubsystem);
 
           // Phase 4 (SSH Console): wire the registry into agent-loop so
           // buildSystemPrompt can auto-inject the live screen tail into
@@ -751,6 +757,21 @@ export function createCoreServer(opts: CoreServerOptions): {
             }),
           );
 
+          // REST user-initiated session-open endpoint:
+          // POST /api/local/tasks/:taskId/console/session. Reuses the same
+          // SshSubsystem + preflight the console tools use; the access gate
+          // runs inside openConsoleSession against task.pieceName.
+          app.use(
+            '/api',
+            express.json(),
+            createConsoleSessionRouter({
+              sub: sshSubsystem,
+              preflight: sshPreflight,
+              requireAuth: authActive ? requireAuth : (_req: Request, _res: Response, next: NextFunction) => next(),
+              resolveTask: consoleDeps.resolveTask,
+            }),
+          );
+
           // Phase 6 (SSH Console): admin list + kill endpoints. The
           // `/api/admin` prefix already has `express.json()` mounted above
           // (see Admin user management API), so POST bodies parse correctly.

src/engine/tools/ssh-console.ts

@@ -32,6 +32,319 @@ import { logger } from '../../logger.js';
 import { getSshSubsystem, preflight, type SshSubsystem } from './ssh.js';
 import { makeNonce, makeMarkerCommand, parseMarker, extractOutput, detectWaitingForInput, shouldGuardInterrupt } from './console-run-lib.js';
 
+// ──────────────────────────────────────────────────────────────────────
+// openConsoleSession — the shared find-or-open core
+// ──────────────────────────────────────────────────────────────────────
+//
+// Extracted from the SshConsoleEnsure tool so a future HTTP endpoint can
+// create a session WITHOUT fabricating a ToolContext. The tool wrapper
+// (ensureSessionInternal) builds OpenConsoleParams from its ToolContext
+// (initiator: 'agent') and maps the structured result back to the tool's
+// existing return/error shape. Every preflight gate, host-key check,
+// disabled/abuse check, key decryption, shell-channel open, session
+// register, per-connection cap, and the `ssh.console.open` audit are
+// preserved verbatim — in the same order, with the same semantics.
+
+/** Explicit collaborators the find-or-open core needs. `sub` carries the
+ * sessionRegistry, connectionRepo, access resolver, key decryptors, audit
+ * repo, abuse repo, config (cols/rows defaults + caps) and openShellChannel.
+ * `preflight` is threaded explicitly (it is the shared access/state gate). */
+export interface OpenConsoleDeps {
+  sub: SshSubsystem;
+  preflight: typeof preflight;
+}
+
+export interface OpenConsoleParams {
+  taskId: string;
+  connectionId: string;
+  /** Connection-resolution / audit owner (job.ownerId ?? 'local' upstream). */
+  ownerId: string | null;
+  /** The acting principal (startedByUserId). */
+  userId: string;
+  /** For the grant check (access resolver matches against this piece). */
+  pieceName: string;
+  /** Piece allowed_ssh_connections (['*'] for user-initiated). */
+  allowedConnections: string[];
+  cols?: number;
+  rows?: number;
+  forceReplace?: boolean;
+  /** Audit marker — distinguishes human-opened from agent-opened sessions. */
+  initiator: 'agent' | 'user';
+  /** Optional job id for the audit rows (agent path passes ctx.jobId). */
+  jobId?: string | null;
+}
+
+export interface OpenConsoleResult {
+  ok: boolean;
+  /** True if a live session already existed and was reused (no open). */
+  alreadyActive?: boolean;
+  connectionId?: string;
+  cols?: number;
+  rows?: number;
+  /** Structured error code on failure. */
+  error?:
+    | 'no_grant'
+    | 'host_key_not_verified'
+    | 'host_key_mismatch'
+    | 'connection_change_requires_force'
+    | 'abuse_locked'
+    | 'connection_disabled'
+    | 'connection_not_found'
+    | 'missing_connection_id'
+    | 'missing_task_context'
+    | 'no_user_context'
+    | 'piece_not_configured'
+    | 'piece_not_allowed'
+    | 'preflight_denied'
+    | 'decrypt_failed'
+    | 'open_shell_failed';
+  /** Human/LLM-readable message (mirrors the tool's existing error strings). */
+  message?: string;
+  /** The live session on success (consumed by the tool wrapper; not serialized). */
+  session?: ConsoleSession;
+}
+
+/**
+ * Find-or-open a ConsoleSession bound to (taskId, connectionId). Runs the
+ * full preflight (piece membership, access decision, enabled / abuse /
+ * host-key state), decrypts key material, opens the shell channel, builds +
+ * registers the ConsoleSession, enforces the per-connection cap, and writes
+ * the `ssh.console.open` audit (with `initiator` in the detail).
+ *
+ * Returns a structured OpenConsoleResult. On the happy path / reuse,
+ * `ok: true` with `session` set; on any gate failure, `ok: false` with a
+ * structured `error` code + `message`.
+ */
+export async function openConsoleSession(
+  deps: OpenConsoleDeps,
+  params: OpenConsoleParams,
+): Promise<OpenConsoleResult> {
+  const { sub } = deps;
+  const connectionId = params.connectionId;
+  if (!connectionId) {
+    return { ok: false, error: 'missing_connection_id', message: 'SshConsoleEnsure: connection_id is required.' };
+  }
+  const localTaskId = params.taskId ?? '';
+  if (!localTaskId) {
+    return {
+      ok: false,
+      error: 'missing_task_context',
+      message: 'SshConsoleEnsure: this tool requires a local task context (ctx.taskId).',
+    };
+  }
+
+  // If a session already exists for this task, branch on whether it's the
+  // same connection. Same → reuse. Different → reject by default (so a
+  // single LLM connection_id slip can't kill the user's live shell), opt
+  // into the swap with force_replace=true.
+  const existing = sub.sessionRegistry.get(localTaskId);
+  if (existing) {
+    if (existing.connectionId === connectionId) {
+      return {
+        ok: true,
+        alreadyActive: true,
+        connectionId: existing.connectionId,
+        cols: existing.cols,
+        rows: existing.rows,
+        session: existing,
+      };
+    }
+    const forceReplace = params.forceReplace === true;
+    if (!forceReplace) {
+      const ageSec = Math.max(0, Math.floor((Date.now() - existing.startedAt) / 1000));
+      const idleSec = Math.max(0, Math.floor((Date.now() - existing.lastActivityAt) / 1000));
+      return {
+        ok: false,
+        error: 'connection_change_requires_force',
+        connectionId: existing.connectionId,
+        message:
+          `SshConsoleEnsure: this task already has an active session on connection ${existing.connectionId} ` +
+          `(age=${ageSec}s, last_activity=${idleSec}s ago). ` +
+          `Use connection_id="${existing.connectionId}" to continue working in the existing shell, ` +
+          `or pass force_replace=true to close it and open a new session on ${connectionId}.`,
+      };
+    }
+    await sub.sessionRegistry.closeForTask(localTaskId, 'connection_change');
+  }
+
+  // Build the minimal ToolContext-shaped object the shared `preflight`
+  // helper reads (userId, ownerId, pieceName, allowedSshConnections, jobId).
+  // The HTTP caller never constructs a ToolContext — params are explicit and
+  // this synthesis is internal to the shared core.
+  const preCtx = {
+    workspacePath: '',
+    editAllowed: false,
+    taskId: localTaskId,
+    userId: params.userId,
+    ownerId: params.ownerId,
+    jobId: params.jobId ?? undefined,
+    pieceName: params.pieceName,
+    allowedSshConnections: params.allowedConnections,
+  } as ToolContext;
+
+  // Full 12-step preflight (same path as SshExec).
+  const pre = deps.preflight({
+    toolName: 'SshExec',
+    connectionId,
+    ctx: preCtx,
+    sub,
+    auditAction: 'ssh.console.open',
+  });
+  if (!pre.ok) {
+    return {
+      ok: false,
+      error: 'preflight_denied',
+      message: pre.error.output,
+    };
+  }
+  const { connection, actingUserId, pieceName } = pre;
+
+  // Console requires a verified host key — there is no LLM-actionable
+  // recovery from first_observe / mismatch on a long-lived shell.
+  if (connection.hostKeyVerifiedAt === null) {
+    sub.auditRepo.beginAndComplete(
+      {
+        action: 'ssh.console.open',
+        connectionId,
+        ownerId: connection.ownerId,
+        actingUserId,
+        pieceName,
+        jobId: preCtx.jobId ?? undefined,
+        detail: { reason: 'host_key_not_verified', initiator: params.initiator },
+      },
+      'denied',
+    );
+    return {
+      ok: false,
+      error: 'host_key_not_verified',
+      message: `SshConsoleEnsure: host key for connection ${connectionId} is not user-verified. Run SshExec first to surface the verify prompt.`,
+    };
+  }
+
+  const cols = typeof params.cols === 'number' && params.cols > 0 ? Math.floor(params.cols) : sub.config.console.defaultCols;
+  const rows = typeof params.rows === 'number' && params.rows > 0 ? Math.floor(params.rows) : sub.config.console.defaultRows;
+
+  // Decrypt key material — same flow as SshExec; we clear on failure but
+  // keep alive past this call because the ssh2 Client needs the PEM through
+  // the entire session. ConsoleSession.close() does NOT clear these
+  // buffers (it can't see them) — we accept that the PEM stays in memory
+  // for the lifetime of the session, which already holds the decrypted
+  // channel and host connection state.
+  let pemBuf: Buffer | null = null;
+  let passBuf: Buffer | null = null;
+  try {
+    pemBuf = sub.decryptKeyMaterial(connection.ownerId, connection.privateKeyEnc);
+    passBuf = sub.decryptPassphrase(connection.ownerId, connection.passphraseEnc);
+  } catch (e) {
+    if (pemBuf) clearBuffer(pemBuf);
+    if (passBuf) clearBuffer(passBuf);
+    sub.auditRepo.beginAndComplete(
+      {
+        action: 'ssh.console.open',
+        connectionId,
+        ownerId: connection.ownerId,
+        actingUserId,
+        pieceName,
+        jobId: preCtx.jobId ?? undefined,
+        detail: { reason: 'decrypt_failed', msg: (e as Error).message, initiator: params.initiator },
+      },
+      'failed',
+    );
+    return { ok: false, error: 'decrypt_failed', message: 'SshConsoleEnsure: failed to decrypt stored key material.' };
+  }
+
+  // Open the channel. On failure clear the PEM and bail.
+  let channel: import('ssh2').ClientChannel;
+  let client: import('ssh2').Client;
+  let hostFingerprint: string;
+  try {
+    const shellResult = await sub.openShellChannel({
+      connection: {
+        id: connection.id,
+        ownerId: connection.ownerId,
+        host: connection.host,
+        port: connection.port,
+        username: connection.username,
+        privateKeyPem: pemBuf,
+        passphrase: passBuf ?? undefined,
+        hostKeyB64: connection.hostKeyB64,
+        hostKeyVerified: true,
+        allowPrivate: sub.config.allowPrivateAddresses || connection.allowPrivateAddresses,
+      },
+      cols,
+      rows,
+      timeoutMs: sub.config.callTimeoutSeconds * 1000,
+    });
+    channel = shellResult.channel;
+    client = shellResult.client;
+    hostFingerprint = shellResult.hostFingerprint;
+  } catch (e) {
+    clearBuffer(pemBuf);
+    clearBuffer(passBuf);
+    sub.abuseRepo.checkAndRecordFailure({
+      connectionId,
+      ownerId: connection.ownerId,
+      userId: actingUserId,
+      host: connection.host,
+      username: connection.username,
+    });
+    sub.auditRepo.beginAndComplete(
+      {
+        action: 'ssh.console.open',
+        connectionId,
+        ownerId: connection.ownerId,
+        actingUserId,
+        pieceName,
+        jobId: preCtx.jobId ?? undefined,
+        detail: { reason: 'open_shell_failed', msg: (e as Error).message, initiator: params.initiator },
+      },
+      'failed',
+    );
+    return { ok: false, error: 'open_shell_failed', message: `SshConsoleEnsure: failed to open shell channel: ${(e as Error).message}` };
+  }
+
+  // Build the session and register it. From here on the channel + client
+  // + PEM belong to the session; we don't clear them on the happy path.
+  // The session ends the client (and thus releases the PEM-bound
+  // connection) when it closes.
+  const session = new ConsoleSession({
+    localTaskId,
+    connectionId,
+    ownerId: connection.ownerId,
+    startedByUserId: actingUserId,
+    cols,
+    rows,
+    scrollbackCap: sub.config.console.scrollbackBytes,
+    channel,
+    client,
+    auditRepo: sub.auditRepo,
+  });
+  sub.sessionRegistry.register(session);
+  sub.abuseRepo.recordSuccess(connectionId);
+  sub.auditRepo.beginAndComplete(
+    {
+      action: 'ssh.console.open',
+      connectionId,
+      ownerId: connection.ownerId,
+      actingUserId,
+      pieceName,
+      jobId: preCtx.jobId ?? undefined,
+      detail: { cols, rows, host_fingerprint: hostFingerprint, initiator: params.initiator },
+    },
+    'success',
+  );
+
+  // Enforce the per-connection session cap (evict oldest).
+  const evict = sub.sessionRegistry.enforceCap(connectionId);
+  for (const e of evict) {
+    sub.sessionRegistry.closeForTask(e.localTaskId, 'session_cap_evict').catch((err) =>
+      logger.warn(`[ssh-console] evict close error: ${(err as Error).message}`),
+    );
+  }
+
+  return { ok: true, alreadyActive: false, connectionId, cols, rows, session };
+}
+
 // ──────────────────────────────────────────────────────────────────────
 // Tool definitions
 // ──────────────────────────────────────────────────────────────────────
@@ -167,6 +480,11 @@ interface EnsureResult {
  * Internal find-or-open helper. Returns a live ConsoleSession bound to
  * (ctx.taskId, connectionId). Used by SshConsoleEnsure directly and by
  * SshConsoleSend / SshConsoleSnapshot when no session is attached yet.
+ *
+ * Thin wrapper over the shared `openConsoleSession` core: it builds
+ * OpenConsoleParams from the ToolContext (initiator: 'agent') and maps the
+ * structured result back to the tool's existing return/error shape so the
+ * tool's external behavior is byte-for-byte unchanged.
  */
 async function ensureSessionInternal(
   input: Record<string, unknown>,
@@ -174,190 +492,30 @@ async function ensureSessionInternal(
   sub: SshSubsystem,
 ): Promise<EnsureResult | ToolResult> {
   const connectionId = typeof input.connection_id === 'string' ? input.connection_id : '';
-  if (!connectionId) {
-    return err('SshConsoleEnsure: connection_id is required.');
-  }
-  const localTaskId = ctx.taskId ?? '';
-  if (!localTaskId) {
-    return err('SshConsoleEnsure: this tool requires a local task context (ctx.taskId).');
-  }
+  const cols = typeof input.cols === 'number' ? input.cols : undefined;
+  const rows = typeof input.rows === 'number' ? input.rows : undefined;
 
-  // If a session already exists for this task, branch on whether it's the
-  // same connection. Same → reuse. Different → reject by default (so a
-  // single LLM connection_id slip can't kill the user's live shell), opt
-  // into the swap with force_replace=true.
-  const existing = sub.sessionRegistry.get(localTaskId);
-  if (existing) {
-    if (existing.connectionId === connectionId) {
-      return { opened: false, session: existing };
-    }
-    const forceReplace = input.force_replace === true;
-    if (!forceReplace) {
-      const ageSec = Math.max(0, Math.floor((Date.now() - existing.startedAt) / 1000));
-      const idleSec = Math.max(0, Math.floor((Date.now() - existing.lastActivityAt) / 1000));
-      return err(
-        `SshConsoleEnsure: this task already has an active session on connection ${existing.connectionId} ` +
-          `(age=${ageSec}s, last_activity=${idleSec}s ago). ` +
-          `Use connection_id="${existing.connectionId}" to continue working in the existing shell, ` +
-          `or pass force_replace=true to close it and open a new session on ${connectionId}.`,
-      );
-    }
-    await sub.sessionRegistry.closeForTask(localTaskId, 'connection_change');
-  }
-
-  // Full 12-step preflight (same path as SshExec).
-  const pre = preflight({
-    toolName: 'SshExec',
-    connectionId,
-    ctx,
-    sub,
-    auditAction: 'ssh.console.open',
-  });
-  if (!pre.ok) return pre.error;
-  const { connection, actingUserId, pieceName } = pre;
-
-  // Console requires a verified host key — there is no LLM-actionable
-  // recovery from first_observe / mismatch on a long-lived shell.
-  if (connection.hostKeyVerifiedAt === null) {
-    sub.auditRepo.beginAndComplete(
-      {
-        action: 'ssh.console.open',
-        connectionId,
-        ownerId: connection.ownerId,
-        actingUserId,
-        pieceName,
-        jobId: ctx.jobId ?? undefined,
-        detail: { reason: 'host_key_not_verified' },
-      },
-      'denied',
-    );
-    return err(
-      `SshConsoleEnsure: host key for connection ${connectionId} is not user-verified. Run SshExec first to surface the verify prompt.`,
-    );
-  }
-
-  const cols = typeof input.cols === 'number' && input.cols > 0 ? Math.floor(input.cols) : sub.config.console.defaultCols;
-  const rows = typeof input.rows === 'number' && input.rows > 0 ? Math.floor(input.rows) : sub.config.console.defaultRows;
-
-  // Decrypt key material — same flow as SshExec; we clear on failure but
-  // keep alive past this call because the ssh2 Client needs the PEM through
-  // the entire session. ConsoleSession.close() does NOT clear these
-  // buffers (it can't see them) — we accept that the PEM stays in memory
-  // for the lifetime of the session, which already holds the decrypted
-  // channel and host connection state.
-  let pemBuf: Buffer | null = null;
-  let passBuf: Buffer | null = null;
-  try {
-    pemBuf = sub.decryptKeyMaterial(connection.ownerId, connection.privateKeyEnc);
-    passBuf = sub.decryptPassphrase(connection.ownerId, connection.passphraseEnc);
-  } catch (e) {
-    if (pemBuf) clearBuffer(pemBuf);
-    if (passBuf) clearBuffer(passBuf);
-    sub.auditRepo.beginAndComplete(
-      {
-        action: 'ssh.console.open',
-        connectionId,
-        ownerId: connection.ownerId,
-        actingUserId,
-        pieceName,
-        jobId: ctx.jobId ?? undefined,
-        detail: { reason: 'decrypt_failed', msg: (e as Error).message },
-      },
-      'failed',
-    );
-    return err('SshConsoleEnsure: failed to decrypt stored key material.');
-  }
-
-  // Open the channel. On failure clear the PEM and bail.
-  let channel: import('ssh2').ClientChannel;
-  let client: import('ssh2').Client;
-  let hostFingerprint: string;
-  try {
-    const shellResult = await sub.openShellChannel({
-      connection: {
-        id: connection.id,
-        ownerId: connection.ownerId,
-        host: connection.host,
-        port: connection.port,
-        username: connection.username,
-        privateKeyPem: pemBuf,
-        passphrase: passBuf ?? undefined,
-        hostKeyB64: connection.hostKeyB64,
-        hostKeyVerified: true,
-        allowPrivate: sub.config.allowPrivateAddresses || connection.allowPrivateAddresses,
-      },
+  const result = await openConsoleSession(
+    { sub, preflight },
+    {
+      taskId: ctx.taskId ?? '',
+      connectionId,
+      ownerId: ctx.ownerId ?? null,
+      userId: (ctx.userId ?? ctx.ownerId ?? '').toString(),
+      pieceName: ctx.pieceName ?? '',
+      allowedConnections: ctx.allowedSshConnections ?? [],
       cols,
       rows,
-      timeoutMs: sub.config.callTimeoutSeconds * 1000,
-    });
-    channel = shellResult.channel;
-    client = shellResult.client;
-    hostFingerprint = shellResult.hostFingerprint;
-  } catch (e) {
-    clearBuffer(pemBuf);
-    clearBuffer(passBuf);
-    sub.abuseRepo.checkAndRecordFailure({
-      connectionId,
-      ownerId: connection.ownerId,
-      userId: actingUserId,
-      host: connection.host,
-      username: connection.username,
-    });
-    sub.auditRepo.beginAndComplete(
-      {
-        action: 'ssh.console.open',
-        connectionId,
-        ownerId: connection.ownerId,
-        actingUserId,
-        pieceName,
-        jobId: ctx.jobId ?? undefined,
-        detail: { reason: 'open_shell_failed', msg: (e as Error).message },
-      },
-      'failed',
-    );
-    return err(`SshConsoleEnsure: failed to open shell channel: ${(e as Error).message}`);
-  }
-
-  // Build the session and register it. From here on the channel + client
-  // + PEM belong to the session; we don't clear them on the happy path.
-  // The session ends the client (and thus releases the PEM-bound
-  // connection) when it closes.
-  const session = new ConsoleSession({
-    localTaskId,
-    connectionId,
-    ownerId: connection.ownerId,
-    startedByUserId: actingUserId,
-    cols,
-    rows,
-    scrollbackCap: sub.config.console.scrollbackBytes,
-    channel,
-    client,
-    auditRepo: sub.auditRepo,
-  });
-  sub.sessionRegistry.register(session);
-  sub.abuseRepo.recordSuccess(connectionId);
-  sub.auditRepo.beginAndComplete(
-    {
-      action: 'ssh.console.open',
-      connectionId,
-      ownerId: connection.ownerId,
-      actingUserId,
-      pieceName,
+      forceReplace: input.force_replace === true,
+      initiator: 'agent',
       jobId: ctx.jobId ?? undefined,
-      detail: { cols, rows, host_fingerprint: hostFingerprint },
     },
-    'success',
   );
 
-  // Enforce the per-connection session cap (evict oldest).
-  const evict = sub.sessionRegistry.enforceCap(connectionId);
-  for (const e of evict) {
-    sub.sessionRegistry.closeForTask(e.localTaskId, 'session_cap_evict').catch((err) =>
-      logger.warn(`[ssh-console] evict close error: ${(err as Error).message}`),
-    );
+  if (!result.ok || !result.session) {
+    return err(result.message ?? 'SshConsoleEnsure: failed to open session.');
   }
-
-  return { opened: true, session };
+  return { opened: !result.alreadyActive, session: result.session };
 }
 
 async function ensureTool(