feat: add shared ai workspace assistant service

services/ai-workspace-assistant/src/server.mjs

@@ -0,0 +1,904 @@
+import express from "express";
+import { randomUUID, timingSafeEqual } from "node:crypto";
+import { createServer } from "node:http";
+import { Pool } from "pg";
+
+const SUPPORTED_SURFACES = new Set(["engine", "ops", "global"]);
+const SUPPORTED_EXECUTOR_TYPES = new Set(["codex-remote", "ndc-agent-core"]);
+const SUPPORTED_CONNECTION_MODES = new Set(["hub", "direct"]);
+const SUPPORTED_EXECUTOR_STATUSES = new Set(["unknown", "online", "offline", "checking", "error"]);
+const SUPPORTED_THREAD_STATES = new Set(["active", "archived"]);
+const SUPPORTED_MESSAGE_ROLES = new Set(["user", "assistant", "system", "tool"]);
+const SUPPORTED_TOOL_PACKS = new Set(["engine", "ops", "ndc-agent-core", "deploy", "docs"]);
+
+const config = readConfig();
+const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
+const app = express();
+const httpServer = createServer(app);
+
+app.disable("x-powered-by");
+app.use(express.json({ limit: "2mb" }));
+
+app.get("/healthz", asyncRoute(async (_req, res) => {
+  await pool.query("select 1");
+  res.json({
+    ok: true,
+    service: "nodedc-ai-workspace-assistant",
+    database: "ready",
+    internalApiConfigured: Boolean(config.internalAccessToken),
+  });
+}));
+
+app.get("/api/ai-workspace/assistant/v1/executors", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const [executors, settings] = await Promise.all([
+    listExecutors(owner),
+    getOwnerSettings(owner),
+  ]);
+  res.json({ ok: true, owner: publicOwner(owner), selectedExecutorId: settings.selectedExecutorId, executors });
+}));
+
+app.post("/api/ai-workspace/assistant/v1/executors", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const command = sanitizeExecutorCommand(req.body, { partial: false });
+  const executor = await createExecutor(owner, command);
+  if (req.body?.select === true) {
+    await selectExecutor(owner, executor.id);
+  }
+  const settings = await getOwnerSettings(owner);
+  res.status(201).json({ ok: true, owner: publicOwner(owner), selectedExecutorId: settings.selectedExecutorId, executor });
+}));
+
+app.patch("/api/ai-workspace/assistant/v1/executors/:executorId", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const executorId = sanitizeUuid(req.params.executorId, "executorId");
+  const command = sanitizeExecutorCommand(req.body, { partial: true });
+  const executor = await updateExecutor(owner, executorId, command);
+  if (!executor) {
+    res.status(404).json({ ok: false, error: "ai_workspace_executor_not_found" });
+    return;
+  }
+  res.json({ ok: true, owner: publicOwner(owner), executor });
+}));
+
+app.delete("/api/ai-workspace/assistant/v1/executors/:executorId", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const executorId = sanitizeUuid(req.params.executorId, "executorId");
+  const deleted = await deleteExecutor(owner, executorId);
+  if (!deleted) {
+    res.status(404).json({ ok: false, error: "ai_workspace_executor_not_found" });
+    return;
+  }
+  res.json({ ok: true, owner: publicOwner(owner), deleted: true });
+}));
+
+app.post("/api/ai-workspace/assistant/v1/executors/:executorId/select", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const executorId = sanitizeUuid(req.params.executorId, "executorId");
+  const executor = await selectExecutor(owner, executorId);
+  if (!executor) {
+    res.status(404).json({ ok: false, error: "ai_workspace_executor_not_found" });
+    return;
+  }
+  res.json({ ok: true, owner: publicOwner(owner), selectedExecutorId: executor.id, executor });
+}));
+
+app.get("/api/ai-workspace/assistant/v1/threads", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const surface = req.query.surface ? sanitizeSurface(req.query.surface) : null;
+  const limit = sanitizeLimit(req.query.limit, 100, 200);
+  const threads = await listThreads(owner, { surface, limit });
+  res.json({ ok: true, owner: publicOwner(owner), threads });
+}));
+
+app.post("/api/ai-workspace/assistant/v1/threads", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const command = sanitizeThreadCommand(req.body, { partial: false });
+  const thread = await createThread(owner, command);
+  res.status(201).json({ ok: true, owner: publicOwner(owner), thread });
+}));
+
+app.get("/api/ai-workspace/assistant/v1/threads/:threadId", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const threadId = sanitizeUuid(req.params.threadId, "threadId");
+  const thread = await getThread(owner, threadId);
+  if (!thread) {
+    res.status(404).json({ ok: false, error: "ai_workspace_thread_not_found" });
+    return;
+  }
+  res.json({ ok: true, owner: publicOwner(owner), thread });
+}));
+
+app.patch("/api/ai-workspace/assistant/v1/threads/:threadId", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const threadId = sanitizeUuid(req.params.threadId, "threadId");
+  const command = sanitizeThreadCommand(req.body, { partial: true });
+  const thread = await updateThread(owner, threadId, command);
+  if (!thread) {
+    res.status(404).json({ ok: false, error: "ai_workspace_thread_not_found" });
+    return;
+  }
+  res.json({ ok: true, owner: publicOwner(owner), thread });
+}));
+
+app.get("/api/ai-workspace/assistant/v1/threads/:threadId/messages", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const threadId = sanitizeUuid(req.params.threadId, "threadId");
+  const limit = sanitizeLimit(req.query.limit, 200, 1000);
+  const cursor = optionalString(req.query.before);
+  const thread = await getThread(owner, threadId);
+  if (!thread) {
+    res.status(404).json({ ok: false, error: "ai_workspace_thread_not_found" });
+    return;
+  }
+  const messages = await listThreadMessages(owner, threadId, { limit, before: cursor });
+  res.json({ ok: true, owner: publicOwner(owner), threadId, messages });
+}));
+
+app.post("/api/ai-workspace/assistant/v1/threads/:threadId/messages", requireInternalApi, asyncRoute(async (req, res) => {
+  const owner = getRequestOwner(req);
+  const threadId = sanitizeUuid(req.params.threadId, "threadId");
+  const command = sanitizeMessageCommand(req.body);
+  const message = await createThreadMessage(owner, threadId, command);
+  if (!message) {
+    res.status(404).json({ ok: false, error: "ai_workspace_thread_not_found" });
+    return;
+  }
+  res.status(201).json({ ok: true, owner: publicOwner(owner), message });
+}));
+
+app.use((error, _req, res, _next) => {
+  const status = Number(error?.status || 500);
+  const message = error instanceof Error ? error.message : "internal_error";
+  res.status(status >= 400 && status < 600 ? status : 500).json({ ok: false, error: message });
+});
+
+await migrate();
+
+httpServer.listen(config.port, "0.0.0.0", () => {
+  console.log(`NODE.DC AI Workspace Assistant listening on http://0.0.0.0:${config.port}`);
+});
+
+process.on("SIGTERM", shutdown);
+process.on("SIGINT", shutdown);
+
+async function listExecutors(owner) {
+  const result = await pool.query(
+    `select *
+    from ai_workspace_executors
+    where owner_key = $1
+    order by updated_at desc, created_at desc`,
+    [owner.key]
+  );
+  return result.rows.map(toExecutor);
+}
+
+async function createExecutor(owner, command) {
+  const result = await pool.query(
+    `insert into ai_workspace_executors (
+      id,
+      owner_key,
+      owner_user_id,
+      owner_email,
+      name,
+      type,
+      connection_mode,
+      endpoint,
+      workspace_path,
+      agent_port,
+      pairing_code,
+      model,
+      account_label,
+      capabilities,
+      status,
+      status_detail,
+      last_seen_at,
+      metadata
+    ) values (
+      $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
+      $11, $12, $13, $14::jsonb, $15, $16, $17, $18::jsonb
+    )
+    returning *`,
+    [
+      randomUUID(),
+      owner.key,
+      owner.userId,
+      owner.email,
+      command.name,
+      command.type,
+      command.connectionMode,
+      command.endpoint,
+      command.workspacePath,
+      command.agentPort,
+      command.pairingCode,
+      command.model,
+      command.accountLabel,
+      JSON.stringify(command.capabilities),
+      command.status,
+      command.statusDetail,
+      command.lastSeenAt,
+      JSON.stringify(command.metadata),
+    ]
+  );
+  return toExecutor(result.rows[0]);
+}
+
+async function updateExecutor(owner, executorId, command) {
+  const fields = [];
+  const values = [owner.key, executorId];
+  const mappings = {
+    name: "name",
+    type: "type",
+    connectionMode: "connection_mode",
+    endpoint: "endpoint",
+    workspacePath: "workspace_path",
+    agentPort: "agent_port",
+    pairingCode: "pairing_code",
+    model: "model",
+    accountLabel: "account_label",
+    capabilities: "capabilities",
+    status: "status",
+    statusDetail: "status_detail",
+    lastSeenAt: "last_seen_at",
+    metadata: "metadata",
+  };
+
+  for (const [key, column] of Object.entries(mappings)) {
+    if (!Object.hasOwn(command, key)) continue;
+    const value = key === "capabilities" || key === "metadata" ? JSON.stringify(command[key]) : command[key];
+    values.push(value);
+    fields.push(`${column} = $${values.length}${key === "capabilities" || key === "metadata" ? "::jsonb" : ""}`);
+  }
+
+  if (fields.length === 0) {
+    return getExecutor(owner, executorId);
+  }
+
+  values.push(new Date());
+  const result = await pool.query(
+    `update ai_workspace_executors
+    set ${fields.join(", ")}, updated_at = $${values.length}
+    where owner_key = $1 and id = $2
+    returning *`,
+    values
+  );
+  return result.rows[0] ? toExecutor(result.rows[0]) : null;
+}
+
+async function deleteExecutor(owner, executorId) {
+  const client = await pool.connect();
+  try {
+    await client.query("begin");
+    await client.query(
+      `update ai_workspace_owner_settings
+      set selected_executor_id = null, updated_at = now()
+      where owner_key = $1 and selected_executor_id = $2`,
+      [owner.key, executorId]
+    );
+    const result = await client.query(
+      "delete from ai_workspace_executors where owner_key = $1 and id = $2",
+      [owner.key, executorId]
+    );
+    await client.query("commit");
+    return result.rowCount > 0;
+  } catch (error) {
+    await client.query("rollback");
+    throw error;
+  } finally {
+    client.release();
+  }
+}
+
+async function getExecutor(owner, executorId) {
+  const result = await pool.query(
+    "select * from ai_workspace_executors where owner_key = $1 and id = $2",
+    [owner.key, executorId]
+  );
+  return result.rows[0] ? toExecutor(result.rows[0]) : null;
+}
+
+async function selectExecutor(owner, executorId) {
+  const executor = await getExecutor(owner, executorId);
+  if (!executor) return null;
+  await pool.query(
+    `insert into ai_workspace_owner_settings (
+      owner_key,
+      owner_user_id,
+      owner_email,
+      selected_executor_id
+    ) values ($1, $2, $3, $4)
+    on conflict (owner_key) do update set
+      owner_user_id = excluded.owner_user_id,
+      owner_email = excluded.owner_email,
+      selected_executor_id = excluded.selected_executor_id,
+      updated_at = now()`,
+    [owner.key, owner.userId, owner.email, executorId]
+  );
+  return executor;
+}
+
+async function getOwnerSettings(owner) {
+  const result = await pool.query(
+    "select * from ai_workspace_owner_settings where owner_key = $1",
+    [owner.key]
+  );
+  return result.rows[0] ? toOwnerSettings(result.rows[0]) : { ownerKey: owner.key, selectedExecutorId: null };
+}
+
+async function listThreads(owner, { surface, limit }) {
+  const values = [owner.key, limit];
+  const surfaceSql = surface ? "and origin_surface = $3" : "";
+  if (surface) values.push(surface);
+  const result = await pool.query(
+    `select *
+    from ai_workspace_threads
+    where owner_key = $1
+      ${surfaceSql}
+    order by updated_at desc, created_at desc
+    limit $2`,
+    values
+  );
+  return result.rows.map(toThread);
+}
+
+async function createThread(owner, command) {
+  const result = await pool.query(
+    `insert into ai_workspace_threads (
+      id,
+      owner_key,
+      owner_user_id,
+      owner_email,
+      title,
+      origin_surface,
+      active_context,
+      enabled_tool_packs,
+      linked_artifacts,
+      selected_executor_id,
+      lifecycle_state,
+      metadata
+    ) values (
+      $1, $2, $3, $4, $5, $6, $7::jsonb, $8::text[], $9::jsonb, $10, $11, $12::jsonb
+    )
+    returning *`,
+    [
+      randomUUID(),
+      owner.key,
+      owner.userId,
+      owner.email,
+      command.title,
+      command.originSurface,
+      JSON.stringify(command.activeContext),
+      command.enabledToolPacks,
+      JSON.stringify(command.linkedArtifacts),
+      command.selectedExecutorId,
+      command.lifecycleState,
+      JSON.stringify(command.metadata),
+    ]
+  );
+  return toThread(result.rows[0]);
+}
+
+async function getThread(owner, threadId) {
+  const result = await pool.query(
+    "select * from ai_workspace_threads where owner_key = $1 and id = $2",
+    [owner.key, threadId]
+  );
+  return result.rows[0] ? toThread(result.rows[0]) : null;
+}
+
+async function updateThread(owner, threadId, command) {
+  const fields = [];
+  const values = [owner.key, threadId];
+  const mappings = {
+    title: "title",
+    originSurface: "origin_surface",
+    activeContext: "active_context",
+    enabledToolPacks: "enabled_tool_packs",
+    linkedArtifacts: "linked_artifacts",
+    selectedExecutorId: "selected_executor_id",
+    lifecycleState: "lifecycle_state",
+    metadata: "metadata",
+  };
+
+  for (const [key, column] of Object.entries(mappings)) {
+    if (!Object.hasOwn(command, key)) continue;
+    const isJson = key === "activeContext" || key === "linkedArtifacts" || key === "metadata";
+    const isArray = key === "enabledToolPacks";
+    const value = isJson ? JSON.stringify(command[key]) : command[key];
+    values.push(value);
+    fields.push(`${column} = $${values.length}${isJson ? "::jsonb" : isArray ? "::text[]" : ""}`);
+  }
+
+  if (fields.length === 0) {
+    return getThread(owner, threadId);
+  }
+
+  values.push(new Date());
+  const result = await pool.query(
+    `update ai_workspace_threads
+    set ${fields.join(", ")}, updated_at = $${values.length}
+    where owner_key = $1 and id = $2
+    returning *`,
+    values
+  );
+  return result.rows[0] ? toThread(result.rows[0]) : null;
+}
+
+async function listThreadMessages(owner, threadId, { limit, before }) {
+  const values = [owner.key, threadId, limit];
+  const beforeSql = before ? "and m.created_at < $4::timestamptz" : "";
+  if (before) values.push(before);
+  const result = await pool.query(
+    `select m.*
+    from ai_workspace_thread_messages m
+    join ai_workspace_threads t on t.id = m.thread_id
+    where t.owner_key = $1
+      and m.thread_id = $2
+      ${beforeSql}
+    order by m.created_at desc, m.sequence desc
+    limit $3`,
+    values
+  );
+  return result.rows.reverse().map(toThreadMessage);
+}
+
+async function createThreadMessage(owner, threadId, command) {
+  const client = await pool.connect();
+  try {
+    await client.query("begin");
+    const thread = await client.query(
+      "select id from ai_workspace_threads where owner_key = $1 and id = $2 for update",
+      [owner.key, threadId]
+    );
+    if (thread.rowCount === 0) {
+      await client.query("rollback");
+      return null;
+    }
+    const sequenceResult = await client.query(
+      "select coalesce(max(sequence), 0) + 1 as next_sequence from ai_workspace_thread_messages where thread_id = $1",
+      [threadId]
+    );
+    const sequence = Number(sequenceResult.rows[0]?.next_sequence || 1);
+    const inserted = await client.query(
+      `insert into ai_workspace_thread_messages (
+        id,
+        thread_id,
+        sequence,
+        role,
+        content,
+        payload
+      ) values ($1, $2, $3, $4, $5, $6::jsonb)
+      returning *`,
+      [randomUUID(), threadId, sequence, command.role, command.content, JSON.stringify(command.payload)]
+    );
+    await client.query("update ai_workspace_threads set updated_at = now() where id = $1", [threadId]);
+    await client.query("commit");
+    return toThreadMessage(inserted.rows[0]);
+  } catch (error) {
+    await client.query("rollback");
+    throw error;
+  } finally {
+    client.release();
+  }
+}
+
+async function migrate() {
+  await pool.query(`
+    create table if not exists ai_workspace_executors (
+      id uuid primary key,
+      owner_key text not null,
+      owner_user_id text,
+      owner_email text,
+      name text not null,
+      type text not null default 'codex-remote',
+      connection_mode text not null default 'hub',
+      endpoint text,
+      workspace_path text,
+      agent_port integer,
+      pairing_code text,
+      model text,
+      account_label text,
+      capabilities jsonb not null default '{}'::jsonb,
+      status text not null default 'unknown',
+      status_detail text,
+      last_seen_at timestamptz,
+      metadata jsonb not null default '{}'::jsonb,
+      created_at timestamptz not null default now(),
+      updated_at timestamptz not null default now(),
+      constraint ai_workspace_executors_owner_check
+        check (owner_user_id is not null or owner_email is not null),
+      constraint ai_workspace_executors_type_check
+        check (type in ('codex-remote', 'ndc-agent-core')),
+      constraint ai_workspace_executors_connection_mode_check
+        check (connection_mode in ('hub', 'direct')),
+      constraint ai_workspace_executors_status_check
+        check (status in ('unknown', 'online', 'offline', 'checking', 'error'))
+    )
+  `);
+
+  await pool.query(`
+    create table if not exists ai_workspace_owner_settings (
+      owner_key text primary key,
+      owner_user_id text,
+      owner_email text,
+      selected_executor_id uuid references ai_workspace_executors(id) on delete set null,
+      metadata jsonb not null default '{}'::jsonb,
+      created_at timestamptz not null default now(),
+      updated_at timestamptz not null default now(),
+      constraint ai_workspace_owner_settings_owner_check
+        check (owner_user_id is not null or owner_email is not null)
+    )
+  `);
+
+  await pool.query(`
+    create table if not exists ai_workspace_threads (
+      id uuid primary key,
+      owner_key text not null,
+      owner_user_id text,
+      owner_email text,
+      title text not null,
+      origin_surface text not null default 'global',
+      active_context jsonb not null default '{}'::jsonb,
+      enabled_tool_packs text[] not null default '{}'::text[],
+      linked_artifacts jsonb not null default '[]'::jsonb,
+      selected_executor_id uuid references ai_workspace_executors(id) on delete set null,
+      lifecycle_state text not null default 'active',
+      metadata jsonb not null default '{}'::jsonb,
+      created_at timestamptz not null default now(),
+      updated_at timestamptz not null default now(),
+      constraint ai_workspace_threads_owner_check
+        check (owner_user_id is not null or owner_email is not null),
+      constraint ai_workspace_threads_surface_check
+        check (origin_surface in ('engine', 'ops', 'global')),
+      constraint ai_workspace_threads_lifecycle_check
+        check (lifecycle_state in ('active', 'archived'))
+    )
+  `);
+
+  await pool.query(`
+    create table if not exists ai_workspace_thread_messages (
+      id uuid primary key,
+      thread_id uuid not null references ai_workspace_threads(id) on delete cascade,
+      sequence integer not null,
+      role text not null,
+      content text not null default '',
+      payload jsonb not null default '{}'::jsonb,
+      created_at timestamptz not null default now(),
+      constraint ai_workspace_thread_messages_role_check
+        check (role in ('user', 'assistant', 'system', 'tool')),
+      unique (thread_id, sequence)
+    )
+  `);
+
+  await pool.query("create index if not exists ai_workspace_executors_owner_idx on ai_workspace_executors(owner_key, updated_at desc)");
+  await pool.query("create index if not exists ai_workspace_executors_pairing_code_idx on ai_workspace_executors(pairing_code) where pairing_code is not null");
+  await pool.query("create index if not exists ai_workspace_threads_owner_surface_idx on ai_workspace_threads(owner_key, origin_surface, updated_at desc)");
+  await pool.query("create index if not exists ai_workspace_thread_messages_thread_idx on ai_workspace_thread_messages(thread_id, sequence)");
+}
+
+function sanitizeExecutorCommand(payload, { partial }) {
+  const source = isPlainObject(payload) ? payload : {};
+  const command = {};
+
+  if (!partial || Object.hasOwn(source, "name")) {
+    command.name = requireNonEmptyString(source.name, "name");
+  }
+  if (!partial || Object.hasOwn(source, "type")) {
+    command.type = sanitizeEnum(source.type || "codex-remote", SUPPORTED_EXECUTOR_TYPES, "unsupported_executor_type");
+  }
+  if (!partial || Object.hasOwn(source, "connectionMode") || Object.hasOwn(source, "connection_mode")) {
+    command.connectionMode = sanitizeEnum(source.connectionMode || source.connection_mode || "hub", SUPPORTED_CONNECTION_MODES, "unsupported_connection_mode");
+  }
+  copyOptionalString(command, "endpoint", source.endpoint);
+  copyOptionalString(command, "workspacePath", source.workspacePath || source.workspace_path);
+  copyOptionalInteger(command, "agentPort", source.agentPort || source.agent_port, 1, 65535);
+  copyOptionalString(command, "pairingCode", source.pairingCode || source.pairing_code);
+  copyOptionalString(command, "model", source.model);
+  copyOptionalString(command, "accountLabel", source.accountLabel || source.account_label);
+  if (!partial || Object.hasOwn(source, "capabilities")) {
+    command.capabilities = isPlainObject(source.capabilities) ? source.capabilities : {};
+  }
+  if (!partial || Object.hasOwn(source, "status")) {
+    command.status = sanitizeEnum(source.status || "unknown", SUPPORTED_EXECUTOR_STATUSES, "unsupported_executor_status");
+  }
+  copyOptionalString(command, "statusDetail", source.statusDetail || source.status_detail);
+  if (Object.hasOwn(source, "lastSeenAt") || Object.hasOwn(source, "last_seen_at")) {
+    command.lastSeenAt = sanitizeOptionalIso(source.lastSeenAt || source.last_seen_at, "lastSeenAt");
+  }
+  if (!partial || Object.hasOwn(source, "metadata")) {
+    command.metadata = isPlainObject(source.metadata) ? source.metadata : {};
+  }
+
+  return command;
+}
+
+function sanitizeThreadCommand(payload, { partial }) {
+  const source = isPlainObject(payload) ? payload : {};
+  const command = {};
+
+  if (!partial || Object.hasOwn(source, "title")) {
+    command.title = requireNonEmptyString(source.title || "AI Workspace Thread", "title");
+  }
+  if (!partial || Object.hasOwn(source, "originSurface") || Object.hasOwn(source, "origin_surface")) {
+    command.originSurface = sanitizeSurface(source.originSurface || source.origin_surface || "global");
+  }
+  if (!partial || Object.hasOwn(source, "activeContext") || Object.hasOwn(source, "active_context")) {
+    const value = source.activeContext || source.active_context;
+    command.activeContext = isPlainObject(value) ? value : {};
+  }
+  if (!partial || Object.hasOwn(source, "enabledToolPacks") || Object.hasOwn(source, "enabled_tool_packs")) {
+    command.enabledToolPacks = sanitizeToolPacks(source.enabledToolPacks || source.enabled_tool_packs || []);
+  }
+  if (!partial || Object.hasOwn(source, "linkedArtifacts") || Object.hasOwn(source, "linked_artifacts")) {
+    const value = source.linkedArtifacts || source.linked_artifacts;
+    command.linkedArtifacts = Array.isArray(value) ? value.filter(isPlainObject) : [];
+  }
+  if (Object.hasOwn(source, "selectedExecutorId") || Object.hasOwn(source, "selected_executor_id")) {
+    const selectedExecutorId = source.selectedExecutorId || source.selected_executor_id;
+    command.selectedExecutorId = selectedExecutorId ? sanitizeUuid(selectedExecutorId, "selectedExecutorId") : null;
+  }
+  if (!partial || Object.hasOwn(source, "lifecycleState") || Object.hasOwn(source, "lifecycle_state")) {
+    command.lifecycleState = sanitizeEnum(source.lifecycleState || source.lifecycle_state || "active", SUPPORTED_THREAD_STATES, "unsupported_thread_state");
+  }
+  if (!partial || Object.hasOwn(source, "metadata")) {
+    command.metadata = isPlainObject(source.metadata) ? source.metadata : {};
+  }
+
+  return command;
+}
+
+function sanitizeMessageCommand(payload) {
+  const source = isPlainObject(payload) ? payload : {};
+  return {
+    role: sanitizeEnum(source.role || "user", SUPPORTED_MESSAGE_ROLES, "unsupported_message_role"),
+    content: optionalString(source.content) || "",
+    payload: isPlainObject(source.payload) ? source.payload : {},
+  };
+}
+
+function getRequestOwner(req) {
+  const userId = optionalString(req.headers["x-nodedc-user-id"] || req.query.userId);
+  const email = normalizeEmail(req.headers["x-nodedc-user-email"] || req.query.email);
+  if (!userId && !email) {
+    throw badRequest("ai_workspace_owner_required");
+  }
+  return {
+    key: userId ? `user:${userId}` : `email:${email}`,
+    userId,
+    email,
+  };
+}
+
+function publicOwner(owner) {
+  return {
+    ownerKey: owner.key,
+    userId: owner.userId,
+    email: owner.email,
+  };
+}
+
+function toExecutor(row) {
+  return {
+    id: row.id,
+    ownerKey: row.owner_key,
+    ownerUserId: row.owner_user_id,
+    ownerEmail: row.owner_email,
+    name: row.name,
+    type: row.type,
+    connectionMode: row.connection_mode,
+    endpoint: row.endpoint,
+    workspacePath: row.workspace_path,
+    agentPort: row.agent_port,
+    pairingCode: row.pairing_code,
+    model: row.model,
+    accountLabel: row.account_label,
+    capabilities: row.capabilities || {},
+    status: row.status,
+    statusDetail: row.status_detail,
+    lastSeenAt: toIso(row.last_seen_at),
+    metadata: row.metadata || {},
+    createdAt: toIso(row.created_at),
+    updatedAt: toIso(row.updated_at),
+  };
+}
+
+function toOwnerSettings(row) {
+  return {
+    ownerKey: row.owner_key,
+    ownerUserId: row.owner_user_id,
+    ownerEmail: row.owner_email,
+    selectedExecutorId: row.selected_executor_id,
+    metadata: row.metadata || {},
+    createdAt: toIso(row.created_at),
+    updatedAt: toIso(row.updated_at),
+  };
+}
+
+function toThread(row) {
+  return {
+    id: row.id,
+    ownerKey: row.owner_key,
+    ownerUserId: row.owner_user_id,
+    ownerEmail: row.owner_email,
+    title: row.title,
+    originSurface: row.origin_surface,
+    activeContext: row.active_context || {},
+    enabledToolPacks: row.enabled_tool_packs || [],
+    linkedArtifacts: row.linked_artifacts || [],
+    selectedExecutorId: row.selected_executor_id,
+    lifecycleState: row.lifecycle_state,
+    metadata: row.metadata || {},
+    createdAt: toIso(row.created_at),
+    updatedAt: toIso(row.updated_at),
+  };
+}
+
+function toThreadMessage(row) {
+  return {
+    id: row.id,
+    threadId: row.thread_id,
+    sequence: row.sequence,
+    role: row.role,
+    content: row.content,
+    payload: row.payload || {},
+    createdAt: toIso(row.created_at),
+  };
+}
+
+function requireInternalApi(req, res, next) {
+  if (!config.internalAccessToken) {
+    res.status(503).json({ ok: false, error: "ai_workspace_assistant_token_not_configured" });
+    return;
+  }
+
+  const authorization = typeof req.headers.authorization === "string" ? req.headers.authorization : "";
+  const bearerToken = authorization.match(/^Bearer\s+(.+)$/i)?.[1] || "";
+  const headerToken = typeof req.headers["x-nodedc-internal-token"] === "string" ? req.headers["x-nodedc-internal-token"] : "";
+  const requestToken = bearerToken || headerToken;
+
+  if (!safeTokenEquals(requestToken, config.internalAccessToken)) {
+    res.status(401).json({ ok: false, error: "ai_workspace_assistant_unauthorized" });
+    return;
+  }
+
+  next();
+}
+
+function safeTokenEquals(actual, expected) {
+  if (!actual || !expected) return false;
+  const actualBuffer = Buffer.from(actual);
+  const expectedBuffer = Buffer.from(expected);
+  if (actualBuffer.length !== expectedBuffer.length) return false;
+  return timingSafeEqual(actualBuffer, expectedBuffer);
+}
+
+function sanitizeSurface(value) {
+  return sanitizeEnum(value || "global", SUPPORTED_SURFACES, "unsupported_surface");
+}
+
+function sanitizeToolPacks(value) {
+  if (!Array.isArray(value)) return [];
+  return Array.from(new Set(value.map(normalizeKey).filter(Boolean))).filter((toolPack) => {
+    if (!SUPPORTED_TOOL_PACKS.has(toolPack)) throw badRequest("unsupported_tool_pack");
+    return true;
+  });
+}
+
+function sanitizeEnum(value, supported, errorMessage) {
+  const key = normalizeKey(value);
+  if (!supported.has(key)) {
+    throw badRequest(errorMessage);
+  }
+  return key;
+}
+
+function sanitizeLimit(value, fallback, max) {
+  const limit = Number(value || fallback);
+  if (!Number.isFinite(limit)) return fallback;
+  return Math.min(Math.max(Math.trunc(limit), 1), max);
+}
+
+function sanitizeUuid(value, name) {
+  const text = optionalString(value);
+  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(text || "")) {
+    throw badRequest(`${name}_invalid`);
+  }
+  return text;
+}
+
+function sanitizeOptionalIso(value, name) {
+  const text = optionalString(value);
+  if (!text) return null;
+  const date = new Date(text);
+  if (Number.isNaN(date.getTime())) {
+    throw badRequest(`${name}_invalid`);
+  }
+  return date.toISOString();
+}
+
+function copyOptionalString(target, key, value) {
+  if (value === undefined) return;
+  target[key] = optionalString(value);
+}
+
+function copyOptionalInteger(target, key, value, min, max) {
+  if (value === undefined) return;
+  if (value === null || value === "") {
+    target[key] = null;
+    return;
+  }
+  const number = Number(value);
+  if (!Number.isInteger(number) || number < min || number > max) {
+    throw badRequest(`${key}_invalid`);
+  }
+  target[key] = number;
+}
+
+function requireNonEmptyString(value, name) {
+  const text = optionalString(value);
+  if (!text) throw badRequest(`${name}_required`);
+  return text;
+}
+
+function optionalString(value) {
+  if (typeof value !== "string") return null;
+  const text = value.trim();
+  return text ? text : null;
+}
+
+function normalizeEmail(value) {
+  const text = optionalString(value);
+  return text ? text.toLowerCase() : null;
+}
+
+function normalizeKey(value) {
+  return String(value || "").trim().toLowerCase().replace(/_/g, "-");
+}
+
+function isPlainObject(value) {
+  return Boolean(value && typeof value === "object" && !Array.isArray(value));
+}
+
+function toIso(value) {
+  if (!value) return null;
+  return value instanceof Date ? value.toISOString() : new Date(value).toISOString();
+}
+
+function badRequest(message) {
+  const error = new Error(message);
+  error.status = 400;
+  return error;
+}
+
+function asyncRoute(handler) {
+  return (req, res, next) => {
+    Promise.resolve(handler(req, res, next)).catch(next);
+  };
+}
+
+function readConfig() {
+  const databaseUrl =
+    process.env.DATABASE_URL ||
+    process.env.AI_WORKSPACE_ASSISTANT_DATABASE_URL ||
+    "postgres://nodedc_ai_workspace:nodedc_ai_workspace@localhost:5432/nodedc_ai_workspace";
+
+  return {
+    port: Number(process.env.PORT || process.env.AI_WORKSPACE_ASSISTANT_PORT || "18082"),
+    databaseUrl,
+    databasePoolSize: Number(process.env.AI_WORKSPACE_ASSISTANT_DATABASE_POOL_SIZE || "10"),
+    internalAccessToken:
+      process.env.AI_WORKSPACE_ASSISTANT_TOKEN ||
+      process.env.NODEDC_INTERNAL_ACCESS_TOKEN ||
+      process.env.NODEDC_PLATFORM_SERVICE_TOKEN ||
+      "",
+  };
+}
+
+async function shutdown() {
+  try {
+    httpServer.close();
+    await pool.end();
+  } finally {
+    process.exit(0);
+  }
+}