/*
 * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p1-zero-split-durable-realtime.
 * 职责: Cloud runtime durable store。Code Agent trace/session/projection cursor 投影事实必须从同一持久化适配器写入和恢复。
 */
import { createHash, randomUUID } from "node:crypto";

import {
  ENVIRONMENT_DEV,
  ERROR_CODES,
  HwlabProtocolError,
  assertProtocolRecord
} from "../protocol/index.mjs";
import {
  CLOUD_API_SERVICE_ID,
  createProtocolAuditEvent,
  deriveProtocolActorFromMeta
} from "../audit/index.mjs";
import {
  CLOUD_CORE_MIGRATION_ID,
  CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION,
  CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE,
  CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS
} from "./schema.ts";

export const RUNTIME_STORE_KIND = "memory";
export const RUNTIME_STORE_KIND_POSTGRES = "postgres";
export const RUNTIME_DURABLE_ADAPTER_MISSING = "runtime_durable_adapter_missing";
export const RUNTIME_DURABLE_ADAPTER_UNCONFIGURED = "runtime_durable_adapter_unconfigured";
export const RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING = "runtime_durable_adapter_driver_missing";
export const RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED = "runtime_durable_adapter_ssl_blocked";
export const RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED = "runtime_durable_adapter_auth_blocked";
export const RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED = "runtime_durable_adapter_schema_blocked";
export const RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED = "runtime_durable_adapter_migration_blocked";
export const RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED = "runtime_durable_adapter_query_blocked";
export const RUNTIME_DURABILITY_REQUIRED_EVIDENCE = "runtime_adapter_schema_migration_read_query";

export const RUNTIME_ADAPTER_ENV = "HWLAB_CLOUD_RUNTIME_ADAPTER";
export const RUNTIME_DURABLE_ENV = "HWLAB_CLOUD_RUNTIME_DURABLE";
export const RUNTIME_DB_POOL_MAX_ENV = "HWLAB_CLOUD_DB_POOL_MAX";

const requiredPostgresSchema = CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS;

const postgresCountTables = Object.freeze([
  "users",
  "user_sessions",
  "gateway_sessions",
  "box_resources",
  "box_capabilities",
  "hardware_operations",
  "audit_events",
  "evidence_records",
  "agent_sessions",
  "account_workspaces",
  "worker_sessions",
  "agent_trace_events",
  "workbench_projection_state",
  "workbench_sessions",
  "workbench_messages",
  "workbench_parts",
  "workbench_turns",
  "workbench_trace_events",
  "workbench_projection_checkpoints"
]);

const postgresRuntimeReadIndexes = Object.freeze([
  "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_trace_order ON agent_trace_events(trace_id, occurred_at, id)",
  "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_session_order ON agent_trace_events(agent_session_id, occurred_at, id)",
  "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_worker_order ON agent_trace_events(worker_session_id, occurred_at, id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated ON workbench_projection_state(projection_status, next_retry_at, updated_at)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_run_command ON workbench_projection_state(run_id, command_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_session_updated ON workbench_projection_state(session_id, updated_at DESC)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_sessions_updated ON workbench_sessions(updated_at DESC, session_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_sessions_owner_updated ON workbench_sessions(owner_user_id, updated_at DESC, session_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_sessions_status_updated ON workbench_sessions(status, updated_at DESC, session_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_messages_session_updated ON workbench_messages(session_id, updated_at ASC, message_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_messages_trace ON workbench_messages(trace_id, message_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_messages_turn ON workbench_messages(turn_id, message_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_parts_message_index ON workbench_parts(message_id, part_index, part_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_parts_session_trace ON workbench_parts(session_id, trace_id, part_index)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_turns_session_updated ON workbench_turns(session_id, updated_at DESC, turn_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_turns_trace ON workbench_turns(trace_id, turn_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_trace_projected_seq ON workbench_trace_events(trace_id, projected_seq)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_trace_events_session_projected_seq ON workbench_trace_events(session_id, projected_seq)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_status_updated ON workbench_projection_checkpoints(projection_status, updated_at DESC, trace_id)",
  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_session_updated ON workbench_projection_checkpoints(session_id, updated_at DESC, trace_id)"
]);

export function createCloudRuntimeStore(options = {}) {
  return new CloudRuntimeStore(options);
}

export function createConfiguredCloudRuntimeStore(options = {}) {
  const env = options.env ?? process.env;
  const adapter = normalizeRuntimeAdapter(options.adapter ?? env?.[RUNTIME_ADAPTER_ENV]);
  const durableRequested =
    options.durable === true ||
    env?.[RUNTIME_DURABLE_ENV] === "1" ||
    env?.[RUNTIME_DURABLE_ENV] === "true";

  if (adapter === RUNTIME_STORE_KIND_POSTGRES || durableRequested) {
    return new PostgresCloudRuntimeStore({
      ...options,
      env,
      dbUrl: options.dbUrl ?? env?.HWLAB_CLOUD_DB_URL,
      sslMode: options.sslMode ?? env?.HWLAB_CLOUD_DB_SSL_MODE
    });
  }

  return createCloudRuntimeStore(options);
}

export function buildPostgresPoolConfig({ dbUrl, sslMode = "require", timeoutMs, poolMax, queryTimeoutMs } = {}) {
  const normalizedSslMode = normalizePostgresSslMode(sslMode);
  const normalizedPoolMax = normalizeRuntimePoolMax(poolMax);
  const normalizedQueryTimeoutMs = queryTimeoutMs === undefined || queryTimeoutMs === null || String(queryTimeoutMs).trim() === ""
    ? null
    : normalizeRuntimeTimeoutMs(queryTimeoutMs);
  return {
    connectionString: normalizePostgresConnectionStringForSslMode(dbUrl, normalizedSslMode),
    ssl: postgresSslOption(normalizedSslMode),
    connectionTimeoutMillis: normalizeRuntimeTimeoutMs(timeoutMs),
    ...(normalizedQueryTimeoutMs !== null ? { query_timeout: normalizedQueryTimeoutMs, statement_timeout: normalizedQueryTimeoutMs } : {}),
    ...(normalizedPoolMax !== null ? { max: normalizedPoolMax } : {})
  };
}

const RUNTIME_DB_QUERY_TIMEOUT_MS_ENV = "HWLAB_CLOUD_DB_QUERY_TIMEOUT_MS";
const RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS_ENV = "HWLAB_CLOUD_DB_QUERY_RETRY_MAX_ATTEMPTS";
const RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS_ENV = "HWLAB_CLOUD_DB_QUERY_RETRY_INITIAL_DELAY_MS";
const RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS_ENV = "HWLAB_CLOUD_DB_QUERY_RETRY_MAX_DELAY_MS";
const DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS = 5;
const DEFAULT_RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS = 250;
const DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS = 5_000;

function normalizePositiveInteger(value, fallback) {
  const number = Number(value);
  return Number.isFinite(number) && number > 0 ? Math.trunc(number) : fallback;
}

function delayRuntimeDbQueryRetry(ms) {
  return new Promise((resolve) => setTimeout(resolve, Math.max(0, Math.trunc(Number(ms) || 0))));
}

export class CloudRuntimeStore {
  constructor({ now = () => new Date().toISOString() } = {}) {
    this.now = now;
    this.gatewaySessions = new Map();
    this.boxResources = new Map();
    this.boxCapabilities = new Map();
    this.hardwareOperations = new Map();
    this.auditEvents = new Map();
    this.evidenceRecords = new Map();
    this.agentTraceEvents = new Map();
    this.workbenchProjectionStates = new Map();
    this.workbenchSessions = new Map();
    this.workbenchMessages = new Map();
    this.workbenchParts = new Map();
    this.workbenchTurns = new Map();
    this.workbenchTraceEvents = new Map();
    this.workbenchProjectionCheckpoints = new Map();
  }

  summary() {
    return addRuntimeDurabilityContract({
      adapter: RUNTIME_STORE_KIND,
      durable: false,
      status: "degraded",
      blocker: RUNTIME_DURABLE_ADAPTER_MISSING,
      reason: "L1 runtime writes are process-local only; no DB-backed durable runtime adapter is configured",
      adapterContract: {
        required: "DB-backed durable runtime adapter using HWLAB_CLOUD_DB_URL",
        sourceIssue: "pikasTech/HWLAB#164",
        secretMaterialRequiredInSource: false
      },
      counts: {
        gatewaySessions: this.gatewaySessions.size,
        boxResources: this.boxResources.size,
        boxCapabilities: this.boxCapabilities.size,
        hardwareOperations: this.hardwareOperations.size,
        auditEvents: this.auditEvents.size,
        evidenceRecords: this.evidenceRecords.size,
        agentTraceEvents: this.agentTraceEvents.size,
        workbenchProjectionStates: this.workbenchProjectionStates.size,
        workbenchSessions: this.workbenchSessions.size,
        workbenchMessages: this.workbenchMessages.size,
        workbenchParts: this.workbenchParts.size,
        workbenchTurns: this.workbenchTurns.size,
        workbenchTraceEvents: this.workbenchTraceEvents.size,
        workbenchProjectionCheckpoints: this.workbenchProjectionCheckpoints.size
      }
    });
  }

  registerGatewaySession(params = {}, requestMeta = {}) {
    const now = this.now();
    const input = asObject(params.gatewaySession ?? params, "gatewaySession");
    const gatewaySession = pruneUndefined({
      gatewaySessionId: input.gatewaySessionId || makeId("gws"),
      projectId: requireProtocolId(input, "projectId"),
      serviceId: input.serviceId || "hwlab-gateway",
      gatewayId: input.gatewayId || makeId("gtw"),
      endpoint: input.endpoint,
      status: input.status || "connected",
      environment: ENVIRONMENT_DEV,
      startedAt: input.startedAt || now,
      lastSeenAt: input.lastSeenAt || now,
      stoppedAt: input.stoppedAt,
      labels: normalizeJsonObject(input.labels)
    });

    assertProtocolRecord("gatewaySession", gatewaySession);
    this.gatewaySessions.set(gatewaySession.gatewaySessionId, gatewaySession);

    const auditEvent = this.recordAuditEvent({
      requestMeta,
      params,
      action: "gateway.session.register",
      targetType: "gateway_session",
      targetId: gatewaySession.gatewaySessionId,
      projectId: gatewaySession.projectId,
      gatewaySessionId: gatewaySession.gatewaySessionId,
      outcome: "accepted",
      metadata: {
        gatewayId: gatewaySession.gatewayId,
        serviceId: gatewaySession.serviceId
      }
    });

    return {
      registered: true,
      gatewaySession,
      auditEvent,
      persistence: this.summary()
    };
  }

  registerBoxResource(params = {}, requestMeta = {}) {
    const now = this.now();
    const input = asObject(params.resource ?? params.boxResource ?? params, "boxResource");
    const gatewaySessionId = requireProtocolId(input, "gatewaySessionId");
    const gatewaySession = this.requireGatewaySession(gatewaySessionId);
    const projectId = input.projectId || gatewaySession.projectId;
    const resource = pruneUndefined({
      resourceId: input.resourceId || makeId("res"),
      projectId,
      gatewaySessionId,
      boxId: input.boxId || makeId("box"),
      resourceType: input.resourceType || "simulator_endpoint",
      name: input.name,
      state: input.state || "available",
      environment: ENVIRONMENT_DEV,
      metadata: normalizeJsonObject(input.metadata),
      createdAt: input.createdAt || now,
      updatedAt: input.updatedAt || now
    });

    assertProtocolRecord("boxResource", resource);
    this.boxResources.set(resource.resourceId, resource);

    const auditEvent = this.recordAuditEvent({
      requestMeta,
      params,
      action: "box.resource.register",
      targetType: "box_resource",
      targetId: resource.resourceId,
      projectId: resource.projectId,
      gatewaySessionId: resource.gatewaySessionId,
      outcome: "accepted",
      metadata: {
        boxId: resource.boxId,
        resourceType: resource.resourceType
      }
    });

    return {
      registered: true,
      resource,
      auditEvent,
      persistence: this.summary()
    };
  }

  reportBoxCapabilities(params = {}, requestMeta = {}) {
    const now = this.now();
    const inputs = normalizeCapabilityInputs(params);
    const capabilities = [];

    for (const input of inputs) {
      const resourceId = requireProtocolId(input, "resourceId");
      const resource = this.requireBoxResource(resourceId);
      const capability = pruneUndefined({
        capabilityId: input.capabilityId || makeId("cap"),
        resourceId,
        projectId: input.projectId || resource.projectId,
        name: requireString(input, "name"),
        description: input.description,
        direction: input.direction || "bidirectional",
        valueType: input.valueType || "object",
        unit: input.unit,
        constraints: normalizeJsonObject(input.constraints),
        mutatesState: input.mutatesState ?? true,
        createdAt: input.createdAt || now,
        updatedAt: input.updatedAt || now
      });

      assertProtocolRecord("boxCapability", capability);
      this.boxCapabilities.set(capability.capabilityId, capability);
      capabilities.push(capability);
    }

    const first = capabilities[0];
    const auditEvent = this.recordAuditEvent({
      requestMeta,
      params,
      action: "box.capability.report",
      targetType: "box_capability",
      targetId: first.capabilityId,
      projectId: first.projectId,
      gatewaySessionId: this.boxResources.get(first.resourceId)?.gatewaySessionId,
      outcome: "accepted",
      metadata: {
        capabilityIds: capabilities.map((capability) => capability.capabilityId)
      }
    });

    return {
      reported: true,
      capabilities,
      auditEvent,
      persistence: this.summary()
    };
  }

  requestHardwareOperation(params = {}, requestMeta = {}) {
    const refs = this.requireOperationRefs(params, requestMeta);
    const now = this.now();
    const operation = pruneUndefined({
      operationId: params.operationId || makeId("op"),
      projectId: refs.projectId,
      gatewaySessionId: refs.gatewaySession.gatewaySessionId,
      agentSessionId: params.agentSessionId,
      workerSessionId: params.workerSessionId,
      resourceId: refs.resource.resourceId,
      capabilityId: refs.capability.capabilityId,
      requestedBy: refs.actor.actorId,
      input: normalizeJsonObject(params.input),
      status: "accepted",
      environment: ENVIRONMENT_DEV,
      requestedAt: params.requestedAt || now,
      updatedAt: now
    });

    assertProtocolRecord("hardwareOperation", operation);
    this.hardwareOperations.set(operation.operationId, operation);

    const auditEvent = this.recordAuditEvent({
      requestMeta,
      params,
      action: "hardware.operation.request",
      targetType: "hardware_operation",
      targetId: operation.operationId,
      projectId: operation.projectId,
      gatewaySessionId: operation.gatewaySessionId,
      workerSessionId: operation.workerSessionId,
      operationId: operation.operationId,
      outcome: "accepted",
      metadata: {
        resourceId: operation.resourceId,
        capabilityId: operation.capabilityId
      }
    });

    return {
      accepted: true,
      status: operation.status,
      operationId: operation.operationId,
      operation,
      auditEvent,
      persistence: this.summary()
    };
  }

  invokeHardwareShell(params = {}, requestMeta = {}) {
    const shellInput = normalizeShellInput(params);
    const operationResult = this.requestHardwareOperation(
      {
        ...params,
        input: {
          ...normalizeJsonObject(params.input),
          shell: shellInput
        }
      },
      requestMeta
    );
    const operation = {
      ...operationResult.operation,
      output: {
        shellExecuted: false,
        dispatchStatus: "not_connected",
        message: "cloud-api recorded the shell invoke request; no gateway shell adapter is connected in L1"
      },
      updatedAt: this.now()
    };
    assertProtocolRecord("hardwareOperation", operation);
    this.hardwareOperations.set(operation.operationId, operation);

    const evidenceRecord = this.writeEvidenceRecord(
      {
        projectId: operation.projectId,
        operationId: operation.operationId,
        kind: "trace",
        mimeType: "application/json",
        serviceId: CLOUD_API_SERVICE_ID,
        metadata: {
          traceId: requestMeta.traceId,
          gatewaySessionId: operation.gatewaySessionId,
          resourceId: operation.resourceId,
          capabilityId: operation.capabilityId,
          shellExecuted: false,
          dispatchStatus: "not_connected"
        },
        content: {
          operationId: operation.operationId,
          shell: shellInput,
          dispatchStatus: "not_connected"
        }
      },
      requestMeta
    ).evidenceRecord;

    const auditEvent = this.recordAuditEvent({
      requestMeta,
      params,
      action: "hardware.invoke.shell",
      targetType: "hardware_operation",
      targetId: operation.operationId,
      projectId: operation.projectId,
      gatewaySessionId: operation.gatewaySessionId,
      workerSessionId: operation.workerSessionId,
      operationId: operation.operationId,
      outcome: "accepted",
      metadata: {
        evidenceId: evidenceRecord.evidenceId,
        shellExecuted: false,
        dispatchStatus: "not_connected"
      }
    });

    return {
      accepted: true,
      status: operation.status,
      operationId: operation.operationId,
      operation,
      auditEvent,
      evidenceRecord,
      dispatch: operation.output,
      persistence: this.summary()
    };
  }

  writeAuditEvent(params = {}, requestMeta = {}) {
    const eventInput = params.event ?? params.auditEvent ?? params;
    const auditEvent = eventInput.auditId
      ? pruneUndefined({
          ...eventInput,
          environment: ENVIRONMENT_DEV
        })
      : this.buildAuditEventFromInput(eventInput, requestMeta);

    assertProtocolRecord("auditEvent", auditEvent);
    this.auditEvents.set(auditEvent.auditId, auditEvent);

    return {
      written: true,
      auditEvent,
      persistence: this.summary()
    };
  }

  queryAuditEvents(params = {}) {
    const events = [...this.auditEvents.values()].filter((event) => matchesQuery(event, params, [
      "auditId",
      "traceId",
      "projectId",
      "gatewaySessionId",
      "operationId",
      "action",
      "targetId"
    ]));

    return {
      events: limitResults(events, params.limit),
      count: events.length,
      persistence: this.summary()
    };
  }

  writeEvidenceRecord(params = {}, requestMeta = {}) {
    const now = this.now();
    const input = asObject(params.record ?? params.evidenceRecord ?? params, "evidenceRecord");
    const content = input.content ?? input.metadata ?? {};
    const sha256 = input.sha256 || sha256Hex(content);
    const evidenceId = input.evidenceId || makeId("evd");
    const evidenceRecord = pruneUndefined({
      evidenceId,
      projectId: requireProtocolId(input, "projectId"),
      operationId: requireProtocolId(input, "operationId"),
      agentSessionId: input.agentSessionId,
      workerSessionId: input.workerSessionId,
      kind: input.kind || "trace",
      uri: input.uri || `memory://hwlab/evidence/${evidenceId}`,
      mimeType: input.mimeType,
      sha256,
      sizeBytes: input.sizeBytes ?? Buffer.byteLength(stableJson(content), "utf8"),
      serviceId: input.serviceId || CLOUD_API_SERVICE_ID,
      environment: ENVIRONMENT_DEV,
      metadata: {
        ...normalizeJsonObject(input.metadata),
        ...(requestMeta.traceId ? { traceId: requestMeta.traceId } : {})
      },
      createdAt: input.createdAt || now
    });

    assertProtocolRecord("evidenceRecord", evidenceRecord);
    this.evidenceRecords.set(evidenceRecord.evidenceId, evidenceRecord);

    return {
      written: true,
      evidenceRecord,
      persistence: this.summary()
    };
  }

  queryEvidenceRecords(params = {}) {
    const records = [...this.evidenceRecords.values()].filter((record) => matchesQuery(record, params, [
      "evidenceId",
      "projectId",
      "operationId",
      "agentSessionId",
      "workerSessionId",
      "kind",
      "serviceId"
    ]));

    return {
      records: limitResults(records, params.limit),
      count: records.length,
      persistence: this.summary()
    };
  }

  writeAgentTraceEvent(params = {}, requestMeta = {}) {
    const traceEvent = normalizeAgentTraceEvent(params.traceEvent ?? params.event ?? params, requestMeta, this.now());
    this.agentTraceEvents.set(traceEvent.id, traceEvent);
    return {
      written: true,
      traceEvent,
      persistence: this.summary()
    };
  }

  queryAgentTraceEvents(params = {}) {
    const records = [...this.agentTraceEvents.values()].filter((record) => matchesQuery(record, params, [
      "id",
      "traceId",
      "agentSessionId",
      "workerSessionId",
      "level"
    ]));
    records.sort(compareTraceEventRecords);
    return {
      events: limitResults(records.map((record) => record.event), params.limit),
      count: records.length,
      persistence: this.summary()
    };
  }

  writeWorkbenchProjectionState(params = {}, requestMeta = {}) {
    const state = normalizeWorkbenchProjectionState(params.projectionState ?? params.state ?? params, requestMeta, this.now());
    this.workbenchProjectionStates.set(state.traceId, state);
    return {
      written: true,
      projectionState: state,
      persistence: this.summary()
    };
  }

  getWorkbenchProjectionState(params = {}) {
    const traceId = textOr(params.traceId, "");
    const state = traceId ? this.workbenchProjectionStates.get(traceId) ?? null : null;
    return {
      projectionState: state,
      found: Boolean(state),
      persistence: this.summary()
    };
  }

  allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) {
    const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now());
    const existing = [...this.workbenchTraceEvents.values()].find((record) =>
      record.traceId === allocation.traceId && record.sourceEventId === allocation.sourceEventId
    );
    if (existing) {
      return {
        ...allocation,
        projectedSeq: nonNegativeInteger(existing.projectedSeq),
        reused: true,
        persistence: this.summary(),
        valuesPrinted: false
      };
    }
    const eventMax = [...this.workbenchTraceEvents.values()]
      .filter((record) => record.traceId === allocation.traceId)
      .reduce((max, record) => Math.max(max, nonNegativeInteger(record.projectedSeq)), 0);
    const checkpointMax = nonNegativeInteger(this.workbenchProjectionCheckpoints.get(allocation.traceId)?.projectedSeq);
    return {
      ...allocation,
      projectedSeq: Math.max(eventMax, checkpointMax) + 1,
      reused: false,
      persistence: this.summary(),
      valuesPrinted: false
    };
  }

  queryWorkbenchProjectionStates(params = {}) {
    const dueAt = textOr(params.dueAt, "");
    const statuses = Array.isArray(params.projectionStatuses) ? new Set(params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean)) : null;
    const records = [...this.workbenchProjectionStates.values()]
      .filter((record) => matchesQuery(record, params, ["traceId", "sessionId", "runId", "commandId", "sourceRunId", "sourceCommandId"]))
      .filter((record) => !statuses || statuses.has(record.projectionStatus))
      .filter((record) => !dueAt || !record.nextRetryAt || String(record.nextRetryAt) <= dueAt)
      .sort(compareProjectionStateRecords);
    return {
      states: limitResults(records, params.limit),
      count: records.length,
      persistence: this.summary()
    };
  }

  writeWorkbenchFacts(params = {}, requestMeta = {}) {
    const facts = normalizeWorkbenchFacts(params.facts ?? params, requestMeta, this.now());
    for (const fact of facts.sessions) this.workbenchSessions.set(fact.sessionId, fact);
    for (const fact of facts.messages) this.workbenchMessages.set(fact.messageId, fact);
    for (const fact of facts.parts) this.workbenchParts.set(fact.partId, fact);
    for (const fact of facts.turns) this.workbenchTurns.set(fact.turnId, fact);
    for (const fact of facts.traceEvents) this.workbenchTraceEvents.set(fact.id, fact);
    for (const fact of facts.checkpoints) {
      const previous = this.workbenchProjectionCheckpoints.get(fact.traceId) ?? null;
      if (!previous || nonNegativeInteger(fact.projectedSeq) >= nonNegativeInteger(previous.projectedSeq)) {
        this.workbenchProjectionCheckpoints.set(fact.traceId, fact);
      }
    }
    return {
      written: true,
      facts,
      persistence: this.summary()
    };
  }

  queryWorkbenchFacts(params = {}) {
    const families = workbenchFactFamilySet(params);
    const sessions = [...this.workbenchSessions.values()].filter((record) => matchesWorkbenchSessionFactQuery(record, params));
    const messages = [...this.workbenchMessages.values()].filter((record) => matchesQuery(record, params, ["messageId", "sessionId", "turnId", "traceId", "role", "status"]));
    const parts = [...this.workbenchParts.values()].filter((record) => matchesQuery(record, params, ["partId", "messageId", "sessionId", "turnId", "traceId", "partType", "status"]));
    const turns = [...this.workbenchTurns.values()].filter((record) => matchesQuery(record, params, ["turnId", "sessionId", "traceId", "messageId", "status"]));
    const afterProjectedSeq = nonNegativeInteger(params.afterProjectedSeq);
    const traceEvents = [...this.workbenchTraceEvents.values()].filter((record) => matchesQuery(record, params, ["id", "traceId", "sessionId", "turnId", "messageId", "eventType"]) && (afterProjectedSeq <= 0 || nonNegativeInteger(record.projectedSeq) > afterProjectedSeq));
    const checkpoints = [...this.workbenchProjectionCheckpoints.values()].filter((record) => matchesQuery(record, params, ["traceId", "sessionId", "turnId", "runId", "commandId", "projectionStatus", "projectionHealth"]));
    const facts = {
      sessions: families.has("sessions") ? limitResults(sortWorkbenchFactRows(sessions, params, "sessions", compareWorkbenchFactRecords), params.limit) : [],
      messages: families.has("messages") ? limitResults(sortWorkbenchFactRows(messages, params, "messages", compareWorkbenchFactRecords), params.limit) : [],
      parts: families.has("parts") ? limitResults(sortWorkbenchFactRows(parts, params, "parts", compareWorkbenchPartFactRecords), params.limit) : [],
      turns: families.has("turns") ? limitResults(sortWorkbenchFactRows(turns, params, "turns", compareWorkbenchFactRecords), params.limit) : [],
      traceEvents: families.has("traceEvents") ? limitResults(sortWorkbenchFactRows(traceEvents, params, "traceEvents", compareWorkbenchTraceEventFacts), params.limit) : [],
      checkpoints: families.has("checkpoints") ? limitResults(sortWorkbenchFactRows(checkpoints, params, "checkpoints", compareWorkbenchFactRecords), params.limit) : []
    };
    return {
      facts,
      count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0),
      persistence: this.summary()
    };
  }

  recordAuditEvent(input) {
    return this.writeAuditEvent(this.buildAuditEventFromInput(input, input.requestMeta), input.requestMeta).auditEvent;
  }

  buildAuditEventFromInput(input = {}, requestMeta = {}) {
    const actor = deriveProtocolActorFromMeta(requestMeta, input.params?.audit ?? input);
    return createProtocolAuditEvent({
      auditId: input.auditId,
      traceId: requestMeta.traceId || input.traceId,
      actorType: actor.actorType,
      actorId: actor.actorId,
      action: input.action,
      targetType: input.targetType,
      targetId: input.targetId,
      projectId: input.projectId,
      gatewaySessionId: input.gatewaySessionId,
      workerSessionId: input.workerSessionId,
      operationId: input.operationId,
      serviceId: CLOUD_API_SERVICE_ID,
      outcome: input.outcome,
      reason: input.reason || input.params?.audit?.reason,
      metadata: input.metadata,
      occurredAt: input.occurredAt
    });
  }

  requireOperationRefs(params, requestMeta = {}) {
    const projectId = requireProtocolId(params, "projectId");
    const gatewaySession = this.requireGatewaySession(requireProtocolId(params, "gatewaySessionId"));
    const resource = this.requireBoxResource(requireProtocolId(params, "resourceId"));
    const capability = this.requireBoxCapability(requireProtocolId(params, "capabilityId"));

    if (gatewaySession.projectId !== projectId || resource.projectId !== projectId || capability.projectId !== projectId) {
      throw new HwlabProtocolError("hardware operation references must belong to one project", {
        code: ERROR_CODES.invalidParams,
        data: {
          projectId,
          gatewaySessionId: gatewaySession.gatewaySessionId,
          resourceId: resource.resourceId,
          capabilityId: capability.capabilityId
        }
      });
    }
    if (resource.gatewaySessionId !== gatewaySession.gatewaySessionId) {
      throw new HwlabProtocolError("resource is not registered under the requested gateway session", {
        code: ERROR_CODES.resourceLocked,
        data: {
          gatewaySessionId: gatewaySession.gatewaySessionId,
          resourceGatewaySessionId: resource.gatewaySessionId,
          resourceId: resource.resourceId
        }
      });
    }
    if (capability.resourceId !== resource.resourceId) {
      throw new HwlabProtocolError("capability is not registered on the requested resource", {
        code: ERROR_CODES.capabilityUnavailable,
        data: {
          resourceId: resource.resourceId,
          capabilityId: capability.capabilityId
        }
      });
    }

    return {
      projectId,
      gatewaySession,
      resource,
      capability,
      actor: deriveProtocolActorFromMeta(requestMeta, params.audit ?? {})
    };
  }

  requireGatewaySession(gatewaySessionId) {
    const gatewaySession = this.gatewaySessions.get(gatewaySessionId);
    if (!gatewaySession) {
      throw new HwlabProtocolError(`gateway session ${gatewaySessionId} is not registered`, {
        code: ERROR_CODES.sessionNotFound,
        data: { gatewaySessionId }
      });
    }
    return gatewaySession;
  }

  requireBoxResource(resourceId) {
    const resource = this.boxResources.get(resourceId);
    if (!resource) {
      throw new HwlabProtocolError(`box resource ${resourceId} is not registered`, {
        code: ERROR_CODES.capabilityUnavailable,
        data: { resourceId }
      });
    }
    return resource;
  }

  requireBoxCapability(capabilityId) {
    const capability = this.boxCapabilities.get(capabilityId);
    if (!capability) {
      throw new HwlabProtocolError(`capability ${capabilityId} is not reported`, {
        code: ERROR_CODES.capabilityUnavailable,
        data: { capabilityId }
      });
    }
    return capability;
  }
}

export class PostgresCloudRuntimeStore {
  constructor({
    env = process.env,
    dbUrl,
    sslMode = "require",
    now = () => new Date().toISOString(),
    queryClient,
    pgModuleLoader,
    logger = console
  } = {}) {
    this.env = env;
    this.dbUrl = dbUrl;
    this.sslMode = sslMode;
    this.now = now;
    this.queryClient = queryClient;
    this.pgModuleLoader = pgModuleLoader;
    this.logger = logger;
    this.pool = null;
    this.runtimeReadIndexesReady = false;
    this.runtimeReadIndexesReadyPromise = null;
    this.memory = new CloudRuntimeStore({ now });
    this.lastReadiness = this.blockedSummary({
      blocker: dbUrl || queryClient ? RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED : RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
      reason: dbUrl || queryClient
        ? "Postgres runtime adapter is configured but has not completed a live schema query"
        : "Postgres runtime adapter requires HWLAB_CLOUD_DB_URL or an injected query client"
    });
  }

  summary() {
    return {
      ...this.lastReadiness,
      counts: this.lastReadiness.counts ?? this.memory.summary().counts
    };
  }

  async readiness() {
    if (!this.dbUrl && !this.queryClient) {
      this.lastReadiness = this.blockedSummary({
        blocker: RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
        reason: "Postgres runtime adapter is selected but HWLAB_CLOUD_DB_URL is not injected"
      });
      return this.summary();
    }

    let rows;
    try {
      const result = await this.query(
        "SELECT table_name, column_name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ANY($1::text[])",
        [Object.keys(requiredPostgresSchema)]
      );
      rows = Array.isArray(result?.rows) ? result.rows : [];
    } catch (error) {
      this.lastReadiness = this.blockedSummary(classifyRuntimeDbError(error));
      return this.summary();
    }

    const schema = summarizeRuntimeSchema(rows);
    if (!schema.ready) {
      this.lastReadiness = this.blockedSummary({
        blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
        reason: "Postgres runtime adapter connected, but required runtime tables or columns are missing",
        schema,
        connection: {
          queryAttempted: true,
          queryResult: "schema_blocked"
        },
        gates: runtimeGates({
          ssl: readyGate(),
          auth: readyGate(),
          schema: blockedGate({ checked: true, blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED }),
          migration: notCheckedGate(),
          durability: blockedGate({ checked: false, blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED })
        })
      });
      return this.summary();
    }

    let migration;
    try {
      migration = await this.readMigrationReadiness();
    } catch (error) {
      const classified = classifyRuntimeDbError(error);
      const migrationReadBlocker = [
        RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED,
        RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
      ].includes(classified.blocker)
        ? classified.blocker
        : RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED;
      this.lastReadiness = this.blockedSummary({
        blocker: migrationReadBlocker,
        reason: migrationReadBlocker === classified.blocker
          ? classified.reason
          : "Postgres runtime schema is present, but the durable runtime migration ledger could not be proven",
        schema,
        migration: blockedRuntimeMigration({ errorCode: classified.connection?.errorCode }),
        connection: {
          queryAttempted: true,
          queryResult: migrationReadBlocker === classified.blocker
            ? classified.connection?.queryResult
            : "migration_blocked",
          errorCode: classified.connection?.errorCode ?? null
        },
        gates: runtimeGates({
          ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : readyGate(),
          auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
              ? notCheckedGate()
            : readyGate(),
          schema: readyGate(),
          migration: blockedGate({
            checked: true,
            blocker: migrationReadBlocker
          }),
          durability: blockedGate({ checked: false, blocker: migrationReadBlocker })
        })
      });
      return this.summary();
    }

    if (!migration.ready) {
      this.lastReadiness = this.blockedSummary({
        blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED,
        reason: "Postgres runtime schema is present, but the required source migration is not recorded",
        schema,
        migration,
        connection: {
          queryAttempted: true,
          queryResult: "migration_blocked"
        },
        gates: runtimeGates({
          ssl: readyGate(),
          auth: readyGate(),
          schema: readyGate(),
          migration: blockedGate({ checked: true, blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED }),
          durability: blockedGate({ checked: false, blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED })
        })
      });
      return this.summary();
    }

    try {
      await this.ensureRuntimeReadIndexes();
    } catch (error) {
      const classified = classifyRuntimeDbError(error);
      this.lastReadiness = this.blockedSummary({
        ...classified,
        schema,
        migration,
        reason: "Postgres runtime schema is present, but Workbench trace read indexes could not be ensured",
        connection: {
          queryAttempted: true,
          queryResult: classified.connection?.queryResult ?? "index_blocked",
          errorCode: classified.connection?.errorCode ?? null
        },
        gates: runtimeGates({
          ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : readyGate(),
          auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
              ? notCheckedGate()
              : readyGate(),
          schema: readyGate(),
          migration: readyGate(),
          durability: blockedGate({ checked: true, blocker: classified.blocker })
        })
      });
      return this.summary();
    }

    let counts;
    try {
      counts = await this.readCounts();
    } catch (error) {
      const classified = classifyRuntimeDbError(error);
      this.lastReadiness = this.blockedSummary({
        ...classified,
        schema,
        migration,
        connection: {
          queryAttempted: true,
          queryResult: classified.connection?.queryResult ?? "query_blocked",
          errorCode: classified.connection?.errorCode ?? null
        },
        gates: runtimeGates({
          ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : readyGate(),
          auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
            ? blockedGate({ checked: true, blocker: classified.blocker })
            : classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
              ? notCheckedGate()
              : readyGate(),
          schema: readyGate(),
          migration: readyGate(),
          durability: blockedGate({ checked: true, blocker: classified.blocker })
        })
      });
      return this.summary();
    }

    this.lastReadiness = addRuntimeDurabilityContract({
      adapter: RUNTIME_STORE_KIND_POSTGRES,
      durable: true,
      durableRequested: true,
      durableCapable: true,
      ready: true,
      status: "ready",
      blocker: null,
      reason: "Postgres durable runtime adapter completed live schema, migration, and read readiness queries",
      liveRuntimeEvidence: true,
      fixtureEvidence: false,
      connection: {
        queryAttempted: true,
        queryResult: "durable_readiness_ready",
        endpointRedacted: true,
        valueRedacted: true
      },
      schema,
      migration,
      gates: runtimeGates({
        ssl: readyGate(),
        auth: readyGate(),
        schema: readyGate(),
        migration: readyGate(),
        durability: readyGate()
      }),
      safety: runtimeSafety(),
      adapterContract: postgresAdapterContract(),
      counts
    });
    return this.summary();
  }

  async registerGatewaySession(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    const before = snapshotMemory(this.memory);
    const result = this.memory.registerGatewaySession(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async registerBoxResource(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    const input = asObject(params.resource ?? params.boxResource ?? params, "boxResource");
    await this.hydrateGatewaySession(input.gatewaySessionId);
    const before = snapshotMemory(this.memory);
    const result = this.memory.registerBoxResource(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async reportBoxCapabilities(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    const inputs = normalizeCapabilityInputs(params);
    for (const input of inputs) {
      await this.hydrateBoxResource(input.resourceId);
    }
    const before = snapshotMemory(this.memory);
    const result = this.memory.reportBoxCapabilities(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async requestHardwareOperation(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    await this.hydrateOperationRefs(params);
    const before = snapshotMemory(this.memory);
    const result = this.memory.requestHardwareOperation(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async invokeHardwareShell(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    await this.hydrateOperationRefs(params);
    const before = snapshotMemory(this.memory);
    const result = this.memory.invokeHardwareShell(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async writeAuditEvent(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    const before = snapshotMemory(this.memory);
    const result = this.memory.writeAuditEvent(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async queryAuditEvents(params = {}) {
    await this.assertReadyForDurableReads("audit.event.query");
    const result = await this.queryDurableReadRows(
      "audit.event.query",
      "SELECT event_json FROM audit_events ORDER BY timestamp ASC",
      []
    );
    const events = result.rows
      .map((row) => parseJsonColumn(row.event_json, null))
      .filter(Boolean)
      .filter((event) => matchesQuery(event, params, [
        "auditId",
        "traceId",
        "projectId",
        "gatewaySessionId",
        "operationId",
        "action",
        "targetId"
      ]));

    return {
      events: limitResults(events, params.limit),
      count: events.length,
      persistence: this.summary()
    };
  }

  async writeEvidenceRecord(params = {}, requestMeta = {}) {
    await this.assertReadyForWrites();
    const before = snapshotMemory(this.memory);
    const result = this.memory.writeEvidenceRecord(params, requestMeta);
    await this.persistChanges(before);
    return withPersistence(result, this.summary());
  }

  async queryEvidenceRecords(params = {}) {
    await this.assertReadyForDurableReads("evidence.record.query");
    const result = await this.queryDurableReadRows(
      "evidence.record.query",
      "SELECT metadata_json FROM evidence_records ORDER BY created_at ASC",
      []
    );
    const records = result.rows
      .map((row) => parseJsonColumn(row.metadata_json, null))
      .filter(Boolean)
      .filter((record) => matchesQuery(record, params, [
        "evidenceId",
        "projectId",
        "operationId",
        "agentSessionId",
        "workerSessionId",
        "kind",
        "serviceId"
      ]));

    return {
      records: limitResults(records, params.limit),
      count: records.length,
      persistence: this.summary()
    };
  }

  async writeAgentTraceEvent(params = {}, requestMeta = {}) {
    const readiness = this.summary();
    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
      await this.assertReadyForWrites();
    }
    const result = this.memory.writeAgentTraceEvent(params, requestMeta);
    await this.persistAgentTraceEvent(result.traceEvent);
    return withPersistence(result, this.summary());
  }

  async queryAgentTraceEvents(params = {}) {
    await this.assertReadyForDurableReads("agent.trace-events.query");
    const traceId = textOr(params.traceId, "");
    const sessionId = textOr(params.sessionId, "");
    const workerSessionId = textOr(params.workerSessionId, "");
    const level = textOr(params.level, "");
    const clauses = [];
    const queryParams = [];
    if (traceId) {
      queryParams.push(traceId);
      clauses.push(`trace_id = $${queryParams.length}`);
    }
    if (sessionId) {
      queryParams.push(sessionId);
      clauses.push(`agent_session_id = $${queryParams.length}`);
    }
    if (workerSessionId) {
      queryParams.push(workerSessionId);
      clauses.push(`worker_session_id = $${queryParams.length}`);
    }
    if (level) {
      queryParams.push(level);
      clauses.push(`level = $${queryParams.length}`);
    }
    const sql = `SELECT event_json FROM agent_trace_events${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY occurred_at ASC, id ASC`;
    const result = await this.queryDurableReadRows("agent.trace-events.query", sql, queryParams);
    const events = result.rows
      .map((row) => parseJsonColumn(row.event_json, null))
      .filter(Boolean)
      .filter((event) => matchesQuery(event, params, ["traceId", "sessionId", "workerSessionId", "level"]));
    return {
      events: limitResults(events, params.limit),
      count: events.length,
      persistence: this.summary()
    };
  }

  async writeWorkbenchProjectionState(params = {}, requestMeta = {}) {
    const readiness = this.summary();
    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
      await this.assertReadyForWrites();
    }
    const result = this.memory.writeWorkbenchProjectionState(params, requestMeta);
    await this.persistWorkbenchProjectionState(result.projectionState);
    return withPersistence(result, this.summary());
  }

  async getWorkbenchProjectionState(params = {}) {
    const traceId = textOr(params.traceId, "");
    if (!traceId) return withPersistence({ projectionState: null, found: false }, this.summary());
    const result = await this.queryWorkbenchProjectionStates({ traceId, limit: 1 });
    const projectionState = result.states[0] ?? null;
    return withPersistence({ projectionState, found: Boolean(projectionState) }, result.persistence);
  }

  async queryWorkbenchProjectionStates(params = {}) {
    await this.assertReadyForDurableReads("workbench.projection-state.query");
    const clauses = [];
    const queryParams = [];
    const addTextClause = (field, column) => {
      const value = textOr(params[field], "");
      if (!value) return;
      queryParams.push(value);
      clauses.push(`${column} = $${queryParams.length}`);
    };
    addTextClause("traceId", "trace_id");
    addTextClause("sessionId", "session_id");
    addTextClause("runId", "run_id");
    addTextClause("commandId", "command_id");
    const statuses = Array.isArray(params.projectionStatuses) ? params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean) : [];
    if (statuses.length > 0) {
      queryParams.push(statuses);
      clauses.push(`projection_status = ANY($${queryParams.length})`);
    }
    const dueAt = textOr(params.dueAt, "");
    if (dueAt) {
      queryParams.push(dueAt);
      clauses.push(`(next_retry_at IS NULL OR next_retry_at <= $${queryParams.length})`);
    }
    const limit = positiveIntegerOrNull(params.limit);
    const offset = positiveIntegerOrNull(params.offset);
    let sql = `SELECT projection_json FROM workbench_projection_state${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ASC, trace_id ASC`;
    if (limit) {
      queryParams.push(limit);
      sql += ` LIMIT $${queryParams.length}`;
    }
    if (offset) {
      queryParams.push(offset);
      sql += ` OFFSET $${queryParams.length}`;
    }
    const result = await this.queryDurableReadRows("workbench.projection-state.query", sql, queryParams);
    const states = result.rows.map((row) => parseJsonColumn(row.projection_json, null)).filter(Boolean);
    return {
      states,
      count: states.length,
      persistence: this.summary()
    };
  }

  async allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) {
    const readiness = this.summary();
    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
      await this.assertReadyForWrites();
    }
    const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now());
    const result = await this.withDurableTransaction("workbench.projected-seq.allocate", async (client) => {
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${allocation.traceId}`]);
      const existing = await client.query(
        "SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2 ORDER BY projected_seq ASC LIMIT 1",
        [allocation.traceId, allocation.sourceEventId]
      );
      const existingSeq = nonNegativeInteger(existing.rows?.[0]?.projected_seq);
      if (existingSeq > 0) return { projectedSeq: existingSeq, reused: true };
      const maxResult = await client.query(
        "SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq",
        [allocation.traceId]
      );
      const proposedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1;
      const reserve = await client.query(
        "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq + 1, EXCLUDED.projected_seq), updated_at = EXCLUDED.updated_at RETURNING projected_seq",
        [allocation.traceId, allocation.sessionId, allocation.turnId, allocation.runId, allocation.commandId, proposedSeq, allocation.sourceSeq, allocation.sourceEventId, "projecting", "healthy", false, false, stableJson({ allocator: "workbench-projected-seq", valuesRedacted: true }), stableJson({ ...allocation, projectedSeq: proposedSeq, projectionStatus: "projecting", projectionHealth: "healthy", valuesPrinted: false }), allocation.createdAt, allocation.updatedAt]
      );
      return { projectedSeq: nonNegativeInteger(reserve.rows?.[0]?.projected_seq) || proposedSeq, reused: false };
    });
    return {
      ...allocation,
      projectedSeq: result.projectedSeq,
      reused: result.reused,
      persistence: this.summary(),
      valuesPrinted: false
    };
  }

  async writeWorkbenchFacts(params = {}, requestMeta = {}) {
    const readiness = this.summary();
    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
      await this.assertReadyForWrites();
    }
    const result = this.memory.writeWorkbenchFacts(params, requestMeta);
    for (const fact of result.facts.sessions) await this.persistWorkbenchSessionFact(fact);
    for (const fact of result.facts.messages) await this.persistWorkbenchMessageFact(fact);
    for (const fact of result.facts.parts) await this.persistWorkbenchPartFact(fact);
    for (const fact of result.facts.turns) await this.persistWorkbenchTurnFact(fact);
    for (const fact of result.facts.traceEvents) await this.persistWorkbenchTraceEventFact(fact);
    for (const fact of result.facts.checkpoints) await this.persistWorkbenchProjectionCheckpoint(fact);
    for (const fact of result.facts.traceEvents) {
      try {
        await this.persistWorkbenchProjectionOutbox({
        traceId: fact.traceId,
        sessionId: fact.sessionId,
        turnId: fact.turnId,
        messageId: fact.messageId,
        projectedSeq: fact.projectedSeq,
        sourceSeq: fact.sourceSeq,
        sourceEventId: fact.sourceEventId,
        commitType: fact.terminal ? "terminal" : "event",
        terminal: Boolean(fact.terminal),
        sealed: Boolean(fact.sealed),
        payload: { type: fact.eventType, projectedSeq: fact.projectedSeq, valuesRedacted: true },
        createdAt: fact.occurredAt ?? fact.updatedAt ?? this.now()
      });
    } catch (outboxError) {
      console.error(`[workbench-projection] outbox write failed: traceId=${fact.traceId} projectedSeq=${fact.projectedSeq} error=${outboxError?.message ?? outboxError}`);
    }
  }
    for (const fact of result.facts.turns) {
      if (fact.terminal || fact.sealed) {
        try {
          await this.persistWorkbenchProjectionOutbox({
          traceId: fact.traceId,
          sessionId: fact.sessionId,
          turnId: fact.turnId,
          messageId: fact.messageId,
          projectedSeq: fact.projectedSeq,
          sourceSeq: fact.sourceSeq,
          sourceEventId: fact.sourceEventId,
          commitType: "terminal",
          terminal: true,
          sealed: true,
          payload: { status: fact.status, failureKind: fact.failureKind, valuesRedacted: true },
          createdAt: fact.updatedAt ?? this.now()
        });
      } catch (outboxError) {
        console.error(`[workbench-projection] terminal outbox write failed: traceId=${fact.traceId} error=${outboxError?.message ?? outboxError}`);
      }
    }
  }
    return withPersistence(result, this.summary());
  }

  async queryWorkbenchFacts(params = {}) {
    await this.assertReadyForDurableReads("workbench.facts.query");
    const families = workbenchFactFamilySet(params);
    // A single Workbench read can request several fact families. Running those
    // SQL reads in parallel lets one HTTP request occupy the whole Postgres
    // pool, which makes the control+observer pages amplify into 503s under
    // periodic refresh. Keep this read model bounded to one connection per
    // request; the page-level latency is preferable to pool starvation.
    const entries = [];
    entries.push(["sessions", families.has("sessions") ? await this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }) : []]);
    entries.push(["messages", families.has("messages") ? await this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }) : []]);
    entries.push(["parts", families.has("parts") ? await this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }) : []]);
    entries.push(["turns", families.has("turns") ? await this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : []]);
    entries.push(["traceEvents", families.has("traceEvents") ? await this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }) : []]);
    entries.push(["checkpoints", families.has("checkpoints") ? await this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" }) : []]);
    const facts = Object.fromEntries(entries);
    return withPersistence({ facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0) }, this.summary());
  }

  async queryWorkbenchFactRows(method, table, jsonColumn, params = {}, columnMap = {}) {
    const family = workbenchFactFamilyForTable(table);
    const queryParams = [];
    const clauses = [];
    for (const [field, column] of Object.entries(columnMap)) {
      const value = textOr(params[field], "");
      if (value) {
        queryParams.push(value);
        clauses.push(`${column} = $${queryParams.length}`);
      }
      const values = textList(params[`${field}s`]);
      if (values.length > 0) {
        queryParams.push(values);
        clauses.push(`${column} = ANY($${queryParams.length})`);
      }
    }
    const limit = positiveIntegerOrNull(params.limit);
    if (family === "traceEvents") {
      const afterProjectedSeq = nonNegativeInteger(params.afterProjectedSeq);
      if (afterProjectedSeq > 0) {
        queryParams.push(afterProjectedSeq);
        clauses.push(`projected_seq > $${queryParams.length}`);
      }
      let sql = `SELECT ${jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY projected_seq ASC`;
      if (limit) {
        queryParams.push(limit);
        sql += ` LIMIT $${queryParams.length}`;
      }
      const result = await this.queryDurableReadRows(method, sql, queryParams);
      return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
    }
    const orderDirection = workbenchFactOrder(params, family) === "updated_desc" ? "DESC" : "ASC";
    const useSessionSummaryProjection = table === "workbench_sessions" && params.sessionProjection === "summary";
    let sql = `SELECT ${useSessionSummaryProjection ? workbenchSessionSummarySelectClause() : jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ${orderDirection}`;
    if (limit) {
      queryParams.push(limit);
      sql += ` LIMIT $${queryParams.length}`;
    }
    const result = await this.queryDurableReadRows(method, sql, queryParams);
    if (useSessionSummaryProjection) return result.rows.map(workbenchSessionSummaryFactFromRow).filter(Boolean);
    return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
  }

  async hydrateOperationRefs(params = {}) {
    await this.hydrateGatewaySession(params.gatewaySessionId);
    await this.hydrateBoxResource(params.resourceId);
    await this.hydrateBoxCapability(params.capabilityId);
  }

  async hydrateGatewaySession(gatewaySessionId) {
    if (!gatewaySessionId || this.memory.gatewaySessions.has(gatewaySessionId)) return;
    const result = await this.query("SELECT gateway_session_json FROM gateway_sessions WHERE id = $1 LIMIT 1", [
      gatewaySessionId
    ]);
    const record = parseJsonColumn(result.rows?.[0]?.gateway_session_json, null);
    if (record) this.memory.gatewaySessions.set(record.gatewaySessionId, record);
  }

  async hydrateBoxResource(resourceId) {
    if (!resourceId || this.memory.boxResources.has(resourceId)) return;
    const result = await this.query("SELECT resource_json FROM box_resources WHERE id = $1 LIMIT 1", [
      resourceId
    ]);
    const record = parseJsonColumn(result.rows?.[0]?.resource_json, null);
    if (record) {
      await this.hydrateGatewaySession(record.gatewaySessionId);
      this.memory.boxResources.set(record.resourceId, record);
    }
  }

  async hydrateBoxCapability(capabilityId) {
    if (!capabilityId || this.memory.boxCapabilities.has(capabilityId)) return;
    const result = await this.query("SELECT capability_json FROM box_capabilities WHERE id = $1 LIMIT 1", [
      capabilityId
    ]);
    const record = parseJsonColumn(result.rows?.[0]?.capability_json, null);
    if (record) {
      await this.hydrateBoxResource(record.resourceId);
      this.memory.boxCapabilities.set(record.capabilityId, record);
    }
  }

  async persistChanges(before) {
    for (const record of newRecords(this.memory.gatewaySessions, before.gatewaySessions)) {
      await this.persistGatewaySession(record);
    }
    for (const record of newRecords(this.memory.boxResources, before.boxResources)) {
      await this.persistBoxResource(record);
    }
    for (const record of newRecords(this.memory.boxCapabilities, before.boxCapabilities)) {
      await this.persistBoxCapability(record);
    }
    for (const record of changedRecords(this.memory.hardwareOperations, before.hardwareOperations)) {
      await this.persistHardwareOperation(record);
    }
    for (const record of newRecords(this.memory.auditEvents, before.auditEvents)) {
      await this.persistAuditEvent(record);
    }
    for (const record of newRecords(this.memory.evidenceRecords, before.evidenceRecords)) {
      await this.persistEvidenceRecord(record);
    }
    for (const record of newRecords(this.memory.agentTraceEvents, before.agentTraceEvents)) {
      await this.persistAgentTraceEvent(record);
    }
    for (const record of changedRecords(this.memory.workbenchProjectionStates, before.workbenchProjectionStates)) {
      await this.persistWorkbenchProjectionState(record);
    }
    for (const record of changedRecords(this.memory.workbenchSessions, before.workbenchSessions)) {
      await this.persistWorkbenchSessionFact(record);
    }
    for (const record of changedRecords(this.memory.workbenchMessages, before.workbenchMessages)) {
      await this.persistWorkbenchMessageFact(record);
    }
    for (const record of changedRecords(this.memory.workbenchParts, before.workbenchParts)) {
      await this.persistWorkbenchPartFact(record);
    }
    for (const record of changedRecords(this.memory.workbenchTurns, before.workbenchTurns)) {
      await this.persistWorkbenchTurnFact(record);
    }
    for (const record of changedRecords(this.memory.workbenchTraceEvents, before.workbenchTraceEvents)) {
      await this.persistWorkbenchTraceEventFact(record);
    }
    for (const record of changedRecords(this.memory.workbenchProjectionCheckpoints, before.workbenchProjectionCheckpoints)) {
      await this.persistWorkbenchProjectionCheckpoint(record);
    }
  }

  async assertReadyForWrites() {
    const readiness = await this.readiness();
    if (readiness.ready !== true) {
      throw new HwlabProtocolError("Postgres durable runtime adapter is not ready for writes", {
        code: ERROR_CODES.internalError,
        data: {
          adapter: RUNTIME_STORE_KIND_POSTGRES,
          blocker: readiness.blocker,
          status: readiness.status,
          schemaReady: Boolean(readiness.schema?.ready)
        },
        context: {
          result: "failed"
        }
      });
    }
  }

  async assertReadyForDurableReads(method) {
    const current = this.summary();
    if (isReadyForDurableRuntimeRead(current)) return;
    const readiness = await this.readiness();
    if (isReadyForDurableRuntimeRead(readiness)) return;
    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
      throw durableRuntimeReadBlockedError(method, readiness);
    }
  }

  async queryDurableReadRows(method, sql, params = []) {
    try {
      return await this.query(sql, params);
    } catch (error) {
      this.lastReadiness = this.blockedSummary({
        ...classifyRuntimeDbError(error),
        reason: `Postgres durable runtime adapter could not complete ${method} read query`
      });
      throw durableRuntimeReadBlockedError(method, this.summary());
    }
  }

  async persistGatewaySession(record) {
    await this.query(
      "INSERT INTO gateway_sessions (id, project_id, gateway_service_id, status, started_at, ended_at, gateway_session_json) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, gateway_service_id = EXCLUDED.gateway_service_id, status = EXCLUDED.status, started_at = EXCLUDED.started_at, ended_at = EXCLUDED.ended_at, gateway_session_json = EXCLUDED.gateway_session_json",
      [
        record.gatewaySessionId,
        record.projectId,
        record.serviceId,
        record.status,
        record.startedAt,
        record.stoppedAt ?? null,
        stableJson(record)
      ]
    );
  }

  async persistBoxResource(record) {
    await this.query(
      "INSERT INTO box_resources (id, project_id, gateway_session_id, resource_state, labels_json, resource_json, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, gateway_session_id = EXCLUDED.gateway_session_id, resource_state = EXCLUDED.resource_state, labels_json = EXCLUDED.labels_json, resource_json = EXCLUDED.resource_json, updated_at = EXCLUDED.updated_at",
      [
        record.resourceId,
        record.projectId,
        record.gatewaySessionId,
        record.state,
        stableJson(record.metadata ?? {}),
        stableJson(record),
        record.updatedAt
      ]
    );
  }

  async persistBoxCapability(record) {
    await this.query(
      "INSERT INTO box_capabilities (id, box_resource_id, capability_type, capability_json, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET box_resource_id = EXCLUDED.box_resource_id, capability_type = EXCLUDED.capability_type, capability_json = EXCLUDED.capability_json, updated_at = EXCLUDED.updated_at",
      [
        record.capabilityId,
        record.resourceId,
        record.name,
        stableJson(record),
        record.updatedAt
      ]
    );
  }

  async persistHardwareOperation(record) {
    await this.query(
      "INSERT INTO hardware_operations (id, project_id, requested_by, operation_type, operation_json, status, requested_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, requested_by = EXCLUDED.requested_by, operation_type = EXCLUDED.operation_type, operation_json = EXCLUDED.operation_json, status = EXCLUDED.status, requested_at = EXCLUDED.requested_at, updated_at = EXCLUDED.updated_at",
      [
        record.operationId,
        record.projectId,
        record.requestedBy,
        record.input?.shell ? "hardware.invoke.shell" : "hardware.operation.request",
        stableJson(record),
        record.status,
        record.requestedAt,
        record.updatedAt
      ]
    );
  }

  async persistAuditEvent(record) {
    await this.query(
      "INSERT INTO audit_events (id, request_id, actor, source, operation, target, result, timestamp, event_json) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (id) DO UPDATE SET request_id = EXCLUDED.request_id, actor = EXCLUDED.actor, source = EXCLUDED.source, operation = EXCLUDED.operation, target = EXCLUDED.target, result = EXCLUDED.result, timestamp = EXCLUDED.timestamp, event_json = EXCLUDED.event_json",
      [
        record.auditId,
        record.traceId,
        stableJson({ type: record.actorType, id: record.actorId }),
        stableJson({ serviceId: record.serviceId, environment: record.environment }),
        record.action,
        stableJson({ type: record.targetType, id: record.targetId }),
        record.outcome ?? "accepted",
        record.occurredAt,
        stableJson(record)
      ]
    );
  }

  async persistEvidenceRecord(record) {
    await this.query(
      "INSERT INTO evidence_records (id, project_id, operation_id, evidence_type, uri, metadata_json, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, operation_id = EXCLUDED.operation_id, evidence_type = EXCLUDED.evidence_type, uri = EXCLUDED.uri, metadata_json = EXCLUDED.metadata_json, created_at = EXCLUDED.created_at",
      [
        record.evidenceId,
        record.projectId,
        record.operationId,
        record.kind,
        record.uri,
        stableJson(record),
        record.createdAt
      ]
    );
  }

  async persistAgentTraceEvent(record) {
    await this.query(
      "INSERT INTO agent_trace_events (id, trace_id, agent_session_id, worker_session_id, level, message, event_json, occurred_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (id) DO UPDATE SET trace_id = EXCLUDED.trace_id, agent_session_id = EXCLUDED.agent_session_id, worker_session_id = EXCLUDED.worker_session_id, level = EXCLUDED.level, message = EXCLUDED.message, event_json = EXCLUDED.event_json, occurred_at = EXCLUDED.occurred_at",
      [
        record.id,
        record.traceId,
        record.agentSessionId,
        record.workerSessionId,
        record.level,
        record.message,
        stableJson(record.event),
        record.occurredAt
      ]
    );
  }

  async persistWorkbenchProjectionState(record) {
    await this.query(
      "INSERT INTO workbench_projection_state (trace_id, session_id, conversation_id, thread_id, run_id, command_id, last_agentrun_seq, last_projected_seq, upstream_latest_seq, projection_status, projection_health, result_sync_state, last_projected_at, last_result_sync_at, last_error_code, last_error_message, failure_count, next_retry_at, projection_json, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21) ON CONFLICT (trace_id) DO UPDATE SET session_id = EXCLUDED.session_id, conversation_id = EXCLUDED.conversation_id, thread_id = EXCLUDED.thread_id, run_id = EXCLUDED.run_id, command_id = EXCLUDED.command_id, last_agentrun_seq = EXCLUDED.last_agentrun_seq, last_projected_seq = EXCLUDED.last_projected_seq, upstream_latest_seq = EXCLUDED.upstream_latest_seq, projection_status = EXCLUDED.projection_status, projection_health = EXCLUDED.projection_health, result_sync_state = EXCLUDED.result_sync_state, last_projected_at = EXCLUDED.last_projected_at, last_result_sync_at = EXCLUDED.last_result_sync_at, last_error_code = EXCLUDED.last_error_code, last_error_message = EXCLUDED.last_error_message, failure_count = EXCLUDED.failure_count, next_retry_at = EXCLUDED.next_retry_at, projection_json = EXCLUDED.projection_json, updated_at = EXCLUDED.updated_at",
      [
        record.traceId,
        record.sessionId,
        record.conversationId,
        record.threadId,
        record.runId,
        record.commandId,
        record.lastAgentRunSeq,
        record.lastProjectedSeq,
        record.upstreamLatestSeq,
        record.projectionStatus,
        record.projectionHealth,
        record.resultSyncState,
        record.lastProjectedAt,
        record.lastResultSyncAt,
        record.lastErrorCode,
        record.lastErrorMessage,
        record.failureCount,
        record.nextRetryAt,
        stableJson(record),
        record.createdAt,
        record.updatedAt
      ]
    );
  }

  async persistWorkbenchSessionFact(record) {
    await this.query(
      "INSERT INTO workbench_sessions (session_id, owner_user_id, project_id, conversation_id, thread_id, status, last_trace_id, projected_seq, source_seq, source_event_id, terminal, sealed, session_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (session_id) DO UPDATE SET owner_user_id = EXCLUDED.owner_user_id, project_id = EXCLUDED.project_id, conversation_id = EXCLUDED.conversation_id, thread_id = EXCLUDED.thread_id, status = EXCLUDED.status, last_trace_id = EXCLUDED.last_trace_id, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, session_json = EXCLUDED.session_json, updated_at = EXCLUDED.updated_at",
      [record.sessionId, record.ownerUserId, record.projectId, record.conversationId, record.threadId, record.status, record.lastTraceId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
    );
  }

  async persistWorkbenchMessageFact(record) {
    await this.query(
      "INSERT INTO workbench_messages (message_id, session_id, turn_id, trace_id, role, status, projected_seq, source_seq, source_event_id, terminal, sealed, message_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (message_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, role = EXCLUDED.role, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, message_json = EXCLUDED.message_json, updated_at = EXCLUDED.updated_at",
      [record.messageId, record.sessionId, record.turnId, record.traceId, record.role, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
    );
  }

  async persistWorkbenchPartFact(record) {
    await this.query(
      "INSERT INTO workbench_parts (part_id, message_id, session_id, turn_id, trace_id, part_index, part_type, status, projected_seq, source_seq, source_event_id, terminal, sealed, part_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (part_id) DO UPDATE SET message_id = EXCLUDED.message_id, session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, part_index = EXCLUDED.part_index, part_type = EXCLUDED.part_type, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, part_json = EXCLUDED.part_json, updated_at = EXCLUDED.updated_at",
      [record.partId, record.messageId, record.sessionId, record.turnId, record.traceId, record.partIndex, record.partType, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt]
    );
  }

  async persistWorkbenchTurnFact(record) {
    await this.query(
      "INSERT INTO workbench_turns (turn_id, session_id, trace_id, message_id, status, projected_seq, source_seq, source_event_id, terminal, sealed, final_response_json, diagnostic_json, turn_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (turn_id) DO UPDATE SET session_id = EXCLUDED.session_id, trace_id = EXCLUDED.trace_id, message_id = EXCLUDED.message_id, status = EXCLUDED.status, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, final_response_json = EXCLUDED.final_response_json, diagnostic_json = EXCLUDED.diagnostic_json, turn_json = EXCLUDED.turn_json, updated_at = EXCLUDED.updated_at",
      [record.turnId, record.sessionId, record.traceId, record.messageId, record.status, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record.finalResponse), stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
    );
  }

  async persistWorkbenchTraceEventFact(record) {
    await this.query(
      "INSERT INTO workbench_trace_events (id, trace_id, session_id, turn_id, message_id, source_seq, source_event_id, projected_seq, event_type, terminal, sealed, event_json, occurred_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (id) DO UPDATE SET trace_id = EXCLUDED.trace_id, session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, message_id = EXCLUDED.message_id, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projected_seq = EXCLUDED.projected_seq, event_type = EXCLUDED.event_type, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, event_json = EXCLUDED.event_json, occurred_at = EXCLUDED.occurred_at, updated_at = EXCLUDED.updated_at",
      [record.id, record.traceId, record.sessionId, record.turnId, record.messageId, record.sourceSeq, record.sourceEventId, record.projectedSeq, record.eventType, record.terminal, record.sealed, stableJson(record), record.occurredAt, record.updatedAt]
    );
  }

  async persistWorkbenchProjectionOutbox(record) {
    await this.query(
      "INSERT INTO workbench_projection_outbox (trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)",
      [record.traceId, record.sessionId, record.turnId, record.messageId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.commitType ?? "event", Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.createdAt ?? this.now()]
    );
  }

  async readWorkbenchProjectionOutbox({ afterSeq = 0, limit = 100, traceId = null, sessionId = null } = {}) {
    const params = [];
    let where = "outbox_seq > $1";
    params.push(Number(afterSeq) || 0);
    let paramIdx = 2;
    if (traceId) {
      where += ` AND trace_id = $${paramIdx}`;
      params.push(traceId);
      paramIdx += 1;
    }
    if (sessionId) {
      where += ` AND session_id = $${paramIdx}`;
      params.push(sessionId);
      paramIdx += 1;
    }
    params.push(Math.min(Math.max(Number(limit) || 100, 1), 500));
    const result = await this.query(
      `SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE ${where} ORDER BY outbox_seq ASC LIMIT $${paramIdx}`,
      params
    );
    return (result.rows ?? []).map((row) => ({
      outboxSeq: Number(row.outbox_seq),
      traceId: row.trace_id,
      sessionId: row.session_id,
      turnId: row.turn_id,
      messageId: row.message_id,
      projectedSeq: Number(row.projected_seq),
      sourceSeq: Number(row.source_seq),
      sourceEventId: row.source_event_id,
      commitType: row.commit_type,
      terminal: Boolean(row.terminal),
      sealed: Boolean(row.sealed),
      payload: typeof row.payload_json === "string" ? JSON.parse(row.payload_json) : row.payload_json,
      createdAt: row.created_at,
      valuesPrinted: false
    }));
  }

  async persistWorkbenchProjectionCheckpoint(record) {
    await this.query(
      "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.session_id ELSE workbench_projection_checkpoints.session_id END, turn_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.turn_id ELSE workbench_projection_checkpoints.turn_id END, run_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.run_id ELSE workbench_projection_checkpoints.run_id END, command_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.command_id ELSE workbench_projection_checkpoints.command_id END, projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq, EXCLUDED.projected_seq), source_seq = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_seq ELSE workbench_projection_checkpoints.source_seq END, source_event_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_event_id ELSE workbench_projection_checkpoints.source_event_id END, projection_status = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_status ELSE workbench_projection_checkpoints.projection_status END, projection_health = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_health ELSE workbench_projection_checkpoints.projection_health END, terminal = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.terminal ELSE workbench_projection_checkpoints.terminal END, sealed = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.sealed ELSE workbench_projection_checkpoints.sealed END, diagnostic_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.diagnostic_json ELSE workbench_projection_checkpoints.diagnostic_json END, checkpoint_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.checkpoint_json ELSE workbench_projection_checkpoints.checkpoint_json END, updated_at = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.updated_at ELSE workbench_projection_checkpoints.updated_at END",
      [record.traceId, record.sessionId, record.turnId, record.runId, record.commandId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.projectionStatus, record.projectionHealth, record.terminal, record.sealed, stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
    );
  }

  async readCounts() {
    const counts = {};
    for (const table of postgresCountTables) {
      const result = await this.query(`SELECT COUNT(*)::int AS count FROM ${table}`, []);
      counts[toCountKey(table)] = Number(result.rows?.[0]?.count ?? 0);
    }
    return counts;
  }

  async readMigrationReadiness() {
    const result = await this.query(
      `SELECT id, schema_version FROM ${CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE} WHERE id = $1 LIMIT 1`,
      [CLOUD_CORE_MIGRATION_ID]
    );
    const row = result.rows?.[0] ?? null;
    const ready =
      row?.id === CLOUD_CORE_MIGRATION_ID &&
      row?.schema_version === CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION;
    return {
      checked: true,
      ready,
      table: CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE,
      requiredMigrationId: CLOUD_CORE_MIGRATION_ID,
      requiredSchemaVersion: CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION,
      appliedMigrationId: row?.id ?? null,
      appliedSchemaVersion: row?.schema_version ?? null,
      missing: !row
    };
  }

  async ensureRuntimeReadIndexes() {
    if (this.runtimeReadIndexesReady) return;
    if (!this.runtimeReadIndexesReadyPromise) {
      this.runtimeReadIndexesReadyPromise = (async () => {
        for (const sql of postgresRuntimeReadIndexes) {
          await this.query(sql, []);
        }
        this.runtimeReadIndexesReady = true;
      })();
    }
    try {
      await this.runtimeReadIndexesReadyPromise;
    } finally {
      if (!this.runtimeReadIndexesReady) this.runtimeReadIndexesReadyPromise = null;
    }
  }

  recordRuntimeDbOperationFailure(label, error, retry = {}) {
    const classified = classifyRuntimeDbError(error);
    this.lastReadiness = this.blockedSummary({
      ...classified,
      reason: `Postgres durable runtime adapter could not complete ${label}`
    });
    this.logger?.warn?.({
      event: "postgres_runtime_operation_failed",
      operation: label,
      blocker: classified.blocker,
      queryResult: classified.connection?.queryResult ?? "query_blocked",
      errorCode: classified.connection?.errorCode ?? "UNKNOWN",
      retryable: classified.retryable === true,
      transient: classified.transient === true,
      retryAfterMs: classified.retryAfterMs ?? null,
      retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
      retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
      retrying: retry.retrying === true,
      retryDelayMs: Number.isFinite(Number(retry.delayMs)) ? Number(retry.delayMs) : null,
      retryLabel: retry.label ?? null,
      valuesRedacted: true,
      endpointRedacted: true
    });
  }

  resetPostgresPoolAfterTransientFailure(classified, retry = {}) {
    if (classified?.connection?.queryResult !== "connect_timeout") return false;
    const pool = this.pool;
    if (!pool) return false;
    this.pool = null;
    this.logger?.warn?.({
      event: "postgres_runtime_pool_reset_after_transient_failure",
      blocker: classified.blocker,
      queryResult: classified.connection?.queryResult ?? "query_blocked",
      errorCode: classified.connection?.errorCode ?? "UNKNOWN",
      retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
      retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
      retrying: retry.retrying === true,
      valuesRedacted: true,
      endpointRedacted: true
    });
    if (typeof pool.end === "function") {
      Promise.resolve()
        .then(() => pool.end())
        .catch((error) => this.logger?.warn?.({
          event: "postgres_runtime_pool_reset_failed",
          errorCode: error?.code ?? "UNKNOWN",
          valuesRedacted: true,
          endpointRedacted: true
        }));
    }
    return true;
  }

  async query(sql, params = []) {
    const retry = this.runtimeQueryRetryPolicy();
    for (let attempt = 1; attempt <= retry.maxAttempts; attempt += 1) {
      try {
        const client = await this.getQueryClient();
        return await client.query(sql, params);
      } catch (error) {
        const classified = classifyRuntimeDbError(error);
        const retrying = classified.retryable === true && classified.transient === true && attempt < retry.maxAttempts;
        const delayMs = retrying ? Math.min(retry.maxDelayMs, retry.initialDelayMs * (2 ** Math.max(0, attempt - 1))) : null;
        this.recordRuntimeDbOperationFailure("query", error, {
          attempt,
          maxAttempts: retry.maxAttempts,
          retrying,
          delayMs,
          label: `${attempt}/${retry.maxAttempts}`
        });
        this.resetPostgresPoolAfterTransientFailure(classified, {
          attempt,
          maxAttempts: retry.maxAttempts,
          retrying,
          delayMs,
          label: `${attempt}/${retry.maxAttempts}`
        });
        if (!retrying) throw error;
        await delayRuntimeDbQueryRetry(delayMs);
      }
    }
  }

  runtimeQueryRetryPolicy() {
    const env = this.env ?? {};
    return {
      maxAttempts: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS),
      initialDelayMs: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS),
      maxDelayMs: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS)
    };
  }

  async withDurableTransaction(label, fn) {
    try {
      const client = await this.getQueryClient();
      if (typeof client.connect === "function") {
        const tx = await client.connect();
        try {
          await tx.query("BEGIN");
          const result = await fn(tx);
          await tx.query("COMMIT");
          return result;
        } catch (error) {
          try { await tx.query("ROLLBACK"); } catch {}
          throw error;
        } finally {
          tx.release?.();
        }
      }
      await client.query("BEGIN");
      try {
        const result = await fn(client);
        await client.query("COMMIT");
        return result;
      } catch (error) {
        try { await client.query("ROLLBACK"); } catch {}
        throw error;
      }
    } catch (error) {
      this.recordRuntimeDbOperationFailure(label || "transaction", error);
      throw error;
    }
  }

  async getQueryClient() {
    if (this.queryClient) {
      return this.queryClient;
    }

    if (this.pool) {
      return this.pool;
    }

    let pg;
    try {
      pg = await (this.pgModuleLoader ? this.pgModuleLoader() : import("pg"));
    } catch (error) {
      if (error?.code === "ERR_MODULE_NOT_FOUND") {
        const driverError = new Error("Postgres runtime adapter requires the pg package");
        driverError.code = "HWLAB_PG_DRIVER_MISSING";
        throw driverError;
      }
      throw error;
    }

    const Pool = pg.Pool ?? pg.default?.Pool;
    if (typeof Pool !== "function") {
      const driverError = new Error("Postgres runtime adapter could not load pg.Pool");
      driverError.code = "HWLAB_PG_DRIVER_MISSING";
      throw driverError;
    }

    this.pool = new Pool(buildPostgresPoolConfig({
      dbUrl: this.dbUrl,
      sslMode: this.sslMode,
      timeoutMs: this.env?.HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS,
      queryTimeoutMs: this.env?.[RUNTIME_DB_QUERY_TIMEOUT_MS_ENV],
      poolMax: this.env?.[RUNTIME_DB_POOL_MAX_ENV]
    }));
    if (typeof this.pool.on === "function") {
      this.pool.on("error", (error) => this.handlePostgresPoolError(error));
    }
    return this.pool;
  }

  handlePostgresPoolError(error) {
    const classified = classifyRuntimeDbError(error);
    this.lastReadiness = this.blockedSummary(classified);
    this.logger?.warn?.({
      event: "postgres_runtime_pool_error",
      blocker: classified.blocker,
      queryResult: classified.connection?.queryResult ?? "query_blocked",
      errorCode: classified.connection?.errorCode ?? "UNKNOWN",
      retryable: classified.retryable === true,
      transient: classified.transient === true,
      retryAfterMs: classified.retryAfterMs ?? null,
      valuesRedacted: true,
      endpointRedacted: true
    });
  }

  blockedSummary({ blocker, reason, schema, migration, connection, gates }) {
    const blockedSchema = schema ?? {
      ready: false,
      checked: false,
      missingTables: [],
      missingColumns: []
    };
    const blockedMigration = migration ?? notCheckedRuntimeMigration();
    return addRuntimeDurabilityContract({
      adapter: RUNTIME_STORE_KIND_POSTGRES,
      durable: false,
      durableRequested: true,
      durableCapable: false,
      ready: false,
      status: "blocked",
      blocker,
      reason,
      liveRuntimeEvidence: false,
      fixtureEvidence: false,
      connection: {
        queryAttempted: Boolean(connection?.queryAttempted),
        queryResult: connection?.queryResult ?? "not_ready",
        endpointRedacted: true,
        valueRedacted: true,
        errorCode: connection?.errorCode ?? null
      },
      schema: blockedSchema,
      migration: blockedMigration,
      gates: gates ?? runtimeGates({
        ...defaultRuntimeGates({ blocker, connection, schema: blockedSchema, migration: blockedMigration }),
        schema: defaultSchemaGate({ blocker, schema: blockedSchema }),
        migration: defaultMigrationGate({ blocker, migration: blockedMigration }),
        durability: blockedGate({ checked: false, blocker })
      }),
      safety: runtimeSafety(),
      adapterContract: postgresAdapterContract()
    });
  }
}

function isReadyForDurableRuntimeRead(readiness = {}) {
  return readiness.ready === true && readiness.durable === true && readiness.liveRuntimeEvidence === true;
}

function normalizeCapabilityInputs(params) {
  const value = params.capabilities ?? params.capability ?? params;
  if (Array.isArray(value)) {
    if (value.length === 0) {
      throw new HwlabProtocolError("capabilities must not be empty", {
        code: ERROR_CODES.invalidParams
      });
    }
    return value.map((item) => asObject(item, "capability"));
  }
  return [asObject(value, "capability")];
}

function normalizeShellInput(params) {
  const input = normalizeJsonObject(params.input);
  const command = params.command ?? input.command;
  if (command === undefined || command === null || command === "") {
    throw new HwlabProtocolError("hardware.invoke.shell requires input.command or command", {
      code: ERROR_CODES.invalidParams,
      data: {
        required: ["projectId", "gatewaySessionId", "resourceId", "capabilityId", "input.command"]
      }
    });
  }
  if (Array.isArray(command)) {
    return {
      command: command.map((item) => String(item)),
      cwd: input.cwd,
      timeoutMs: input.timeoutMs
    };
  }
  return {
    command: String(command),
    cwd: input.cwd,
    timeoutMs: input.timeoutMs
  };
}

function normalizeAgentTraceEvent(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
  if (!traceId) {
    throw new HwlabProtocolError("agent trace event requires traceId", {
      code: ERROR_CODES.invalidParams,
      data: { required: ["traceId"] }
    });
  }
  const seq = positiveIntegerOrNull(input.seq);
  const occurredAt = timestampOr(input.occurredAt ?? input.createdAt ?? input.timestamp, now);
  const status = textOr(input.status, "observed");
  const type = textOr(input.type ?? input.kind, "event");
  const level = traceEventLevel(input.level ?? status ?? type);
  const message = textOr(input.message ?? input.label ?? type, "");
  const agentSessionId = textOr(input.agentSessionId ?? input.sessionId ?? requestMeta.agentSessionId ?? requestMeta.sessionId, "");
  const workerSessionId = textOr(input.workerSessionId ?? input.workerSession?.id ?? input.workerId, "");
  const id = textOr(input.id ?? input.eventId, "") || `tev_${sha256Hex({
    traceId,
    seq,
    source: input.source ?? null,
    sourceSeq: input.sourceSeq ?? null,
    type,
    status,
    label: input.label ?? null,
    message,
    occurredAt
  }).slice(0, 32)}`;
  const event = pruneUndefined({
    ...input,
    traceId,
    seq: seq ?? undefined,
    agentSessionId: agentSessionId || undefined,
    workerSessionId: workerSessionId || undefined,
    level,
    createdAt: timestampOr(input.createdAt ?? occurredAt, occurredAt),
    valuesPrinted: false
  });
  return {
    id,
    traceId,
    agentSessionId: agentSessionId || null,
    workerSessionId: workerSessionId || null,
    level,
    message,
    occurredAt,
    event
  };
}

function normalizeWorkbenchProjectionState(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
  const runId = textOr(input.runId ?? input.sourceRunId, "");
  const commandId = textOr(input.commandId ?? input.sourceCommandId, "");
  if (!traceId || !runId || !commandId) {
    throw new HwlabProtocolError("workbench projection state requires traceId, runId, and commandId", {
      code: ERROR_CODES.invalidParams,
      data: { required: ["traceId", "runId", "commandId"] }
    });
  }
  const timestamp = timestampOr(input.updatedAt, now);
  const lastAgentRunSeq = nonNegativeInteger(input.lastAgentRunSeq ?? input.lastSourceSeq);
  const lastProjectedSeq = nonNegativeInteger(input.lastProjectedSeq ?? lastAgentRunSeq);
  const upstreamLatestSeq = nonNegativeIntegerOrNull(input.upstreamLatestSeq ?? input.sourceLatestSeq);
  return pruneUndefined({
    ...input,
    traceId,
    sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
    conversationId: textOr(input.conversationId, "") || null,
    threadId: textOr(input.threadId, "") || null,
    ownerUserId: textOr(input.ownerUserId, "") || null,
    ownerRole: textOr(input.ownerRole, "") || null,
    sourceRunId: runId,
    sourceCommandId: commandId,
    runId,
    commandId,
    lastSourceSeq: lastAgentRunSeq,
    lastAgentRunSeq,
    lastProjectedSeq,
    sourceLatestSeq: upstreamLatestSeq ?? lastAgentRunSeq,
    upstreamLatestSeq: upstreamLatestSeq ?? lastAgentRunSeq,
    projectionStatus: projectionStatusOr(input.projectionStatus, "projecting"),
    projectionHealth: projectionHealthOr(input.projectionHealth, "healthy"),
    resultSyncState: resultSyncStateOr(input.resultSyncState, "not_started"),
    lastProjectedAt: timestampOr(input.lastProjectedAt, timestamp),
    lastResultSyncAt: input.lastResultSyncAt ? timestampOr(input.lastResultSyncAt, timestamp) : null,
    lastErrorCode: textOr(input.lastErrorCode, "") || null,
    lastErrorMessage: textOr(input.lastErrorMessage, "") || null,
    failureCount: nonNegativeInteger(input.failureCount),
    nextRetryAt: input.nextRetryAt ? timestampOr(input.nextRetryAt, timestamp) : null,
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchProjectionAllocation(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
  const sourceEventId = textOr(input.sourceEventId ?? input.eventId, "");
  if (!traceId || !sourceEventId) {
    throw new HwlabProtocolError("workbench projectedSeq allocation requires traceId and sourceEventId", {
      code: ERROR_CODES.invalidParams,
      data: { required: ["traceId", "sourceEventId"] }
    });
  }
  const timestamp = timestampOr(input.updatedAt ?? input.occurredAt ?? input.createdAt, now);
  return pruneUndefined({
    ...input,
    traceId,
    sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
    turnId: textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null,
    runId: textOr(input.runId ?? input.sourceRunId, "") || null,
    commandId: textOr(input.commandId ?? input.sourceCommandId, "") || null,
    sourceSeq: nonNegativeInteger(input.sourceSeq ?? input.lastSourceSeq ?? input.lastAgentRunSeq),
    sourceEventId,
    eventType: textOr(input.eventType ?? input.type, "event"),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchFacts(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  return {
    sessions: arrayInput(input.sessions ?? input.session).map((item) => normalizeWorkbenchSessionFact(item, requestMeta, now)),
    messages: arrayInput(input.messages ?? input.message).map((item) => normalizeWorkbenchMessageFact(item, requestMeta, now)),
    parts: arrayInput(input.parts ?? input.part).map((item) => normalizeWorkbenchPartFact(item, requestMeta, now)),
    turns: arrayInput(input.turns ?? input.turn).map((item) => normalizeWorkbenchTurnFact(item, requestMeta, now)),
    traceEvents: arrayInput(input.traceEvents ?? input.traceEvent).map((item) => normalizeWorkbenchTraceEventFact(item, requestMeta, now)),
    checkpoints: arrayInput(input.checkpoints ?? input.checkpoint).map((item) => normalizeWorkbenchProjectionCheckpoint(item, requestMeta, now))
  };
}

function normalizeWorkbenchSessionFact(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const sessionId = textOr(input.sessionId ?? input.id ?? requestMeta.sessionId, "");
  if (!sessionId) throw requiredWorkbenchFactError("workbench session fact", ["sessionId"]);
  const sourceSeq = nonNegativeInteger(input.sourceSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? sourceSeq);
  const timestamp = timestampOr(input.updatedAt ?? input.createdAt, now);
  return pruneUndefined({
    ...input,
    id: sessionId,
    sessionId,
    ownerUserId: textOr(input.ownerUserId ?? requestMeta.ownerUserId, "") || null,
    projectId: textOr(input.projectId ?? requestMeta.projectId, "") || null,
    conversationId: textOr(input.conversationId ?? requestMeta.conversationId, "") || null,
    threadId: textOr(input.threadId ?? requestMeta.threadId, "") || null,
    status: textOr(input.status, "unknown"),
    lastTraceId: textOr(input.lastTraceId ?? input.traceId ?? requestMeta.traceId, "") || null,
    projectedSeq,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchMessageFact(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const sessionId = textOr(input.sessionId ?? requestMeta.sessionId, "");
  if (!sessionId) throw requiredWorkbenchFactError("workbench message fact", ["sessionId"]);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "") || null;
  const turnId = textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null;
  const role = textOr(input.role, "agent");
  const sourceSeq = nonNegativeInteger(input.sourceSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? sourceSeq);
  const timestamp = timestampOr(input.updatedAt ?? input.createdAt, now);
  const messageId = textOr(input.messageId ?? input.id, "") || stableWorkbenchFactId("msg", { sessionId, turnId, traceId, role, sourceSeq, projectedSeq });
  return pruneUndefined({
    ...input,
    id: messageId,
    messageId,
    sessionId,
    turnId,
    traceId,
    role,
    status: textOr(input.status, "unknown"),
    projectedSeq,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchPartFact(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const sessionId = textOr(input.sessionId ?? requestMeta.sessionId, "");
  const messageId = textOr(input.messageId ?? requestMeta.messageId, "");
  if (!sessionId || !messageId) throw requiredWorkbenchFactError("workbench part fact", ["sessionId", "messageId"]);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "") || null;
  const turnId = textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null;
  const partIndex = nonNegativeInteger(input.partIndex ?? input.index);
  const sourceSeq = nonNegativeInteger(input.sourceSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? sourceSeq);
  const timestamp = timestampOr(input.updatedAt ?? input.createdAt, now);
  const partType = textOr(input.partType ?? input.type, "text");
  const partId = textOr(input.partId ?? input.id, "") || stableWorkbenchFactId("wpt", { sessionId, messageId, turnId, traceId, partIndex, partType });
  return pruneUndefined({
    ...input,
    id: partId,
    partId,
    messageId,
    sessionId,
    turnId,
    traceId,
    partIndex,
    partType,
    status: textOr(input.status, "unknown"),
    projectedSeq,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchTurnFact(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const sessionId = textOr(input.sessionId ?? requestMeta.sessionId, "");
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "") || null;
  const turnId = textOr(input.turnId ?? input.id ?? requestMeta.turnId ?? traceId, "");
  if (!sessionId || !turnId) throw requiredWorkbenchFactError("workbench turn fact", ["sessionId", "turnId"]);
  const sourceSeq = nonNegativeInteger(input.sourceSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? sourceSeq);
  const timestamp = timestampOr(input.updatedAt ?? input.createdAt, now);
  return pruneUndefined({
    ...input,
    id: turnId,
    turnId,
    sessionId,
    traceId,
    messageId: textOr(input.messageId ?? requestMeta.messageId, "") || null,
    status: textOr(input.status, "unknown"),
    projectedSeq,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    finalResponse: input.finalResponse ?? null,
    diagnostic: normalizeJsonObject(input.diagnostic ?? input.projection ?? input.blocker),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function normalizeWorkbenchTraceEventFact(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
  if (!traceId) throw requiredWorkbenchFactError("workbench trace event fact", ["traceId"]);
  const sessionId = textOr(input.sessionId ?? requestMeta.sessionId, "") || null;
  const turnId = textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null;
  const sourceSeq = nonNegativeInteger(input.sourceSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? input.seq);
  const occurredAt = timestampOr(input.occurredAt ?? input.createdAt ?? input.timestamp, now);
  const eventType = textOr(input.eventType ?? input.type ?? input.kind, "event");
  const id = textOr(input.id ?? input.eventId, "") || stableWorkbenchFactId("wte", { traceId, sourceSeq, projectedSeq, eventType, sourceEventId: input.sourceEventId ?? null, occurredAt });
  return pruneUndefined({
    ...input,
    id,
    traceId,
    sessionId,
    turnId,
    messageId: textOr(input.messageId ?? requestMeta.messageId, "") || null,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    projectedSeq,
    eventType,
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    occurredAt,
    updatedAt: timestampOr(input.updatedAt, occurredAt),
    valuesPrinted: false
  });
}

function normalizeWorkbenchProjectionCheckpoint(value, requestMeta = {}, now) {
  const input = normalizeJsonObject(value);
  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
  if (!traceId) throw requiredWorkbenchFactError("workbench projection checkpoint", ["traceId"]);
  const sourceSeq = nonNegativeInteger(input.sourceSeq ?? input.lastSourceSeq ?? input.lastAgentRunSeq);
  const projectedSeq = nonNegativeInteger(input.projectedSeq ?? input.lastProjectedSeq ?? sourceSeq);
  const timestamp = timestampOr(input.updatedAt ?? input.lastProjectedAt, now);
  return pruneUndefined({
    ...input,
    id: traceId,
    traceId,
    sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
    turnId: textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null,
    runId: textOr(input.runId ?? input.sourceRunId, "") || null,
    commandId: textOr(input.commandId ?? input.sourceCommandId, "") || null,
    projectedSeq,
    sourceSeq,
    sourceEventId: textOr(input.sourceEventId ?? input.eventId, "") || null,
    projectionStatus: projectionStatusOr(input.projectionStatus, "projecting"),
    projectionHealth: projectionHealthOr(input.projectionHealth, "healthy"),
    terminal: Boolean(input.terminal),
    sealed: Boolean(input.sealed),
    diagnostic: normalizeJsonObject(input.diagnostic ?? input.blocker),
    createdAt: timestampOr(input.createdAt, timestamp),
    updatedAt: timestamp,
    valuesPrinted: false
  });
}

function arrayInput(value) {
  if (Array.isArray(value)) return value;
  if (value && typeof value === "object") return [value];
  return [];
}

function stableWorkbenchFactId(prefix, value) {
  return `${prefix}_${sha256Hex(value).slice(0, 32)}`;
}

function requiredWorkbenchFactError(label, fields) {
  return new HwlabProtocolError(`${label} requires ${fields.join(", ")}`, {
    code: ERROR_CODES.invalidParams,
    data: { required: fields }
  });
}

function compareTraceEventRecords(left, right) {
  const leftSeq = positiveIntegerOrNull(left?.event?.seq);
  const rightSeq = positiveIntegerOrNull(right?.event?.seq);
  if (leftSeq && rightSeq && leftSeq !== rightSeq) return leftSeq - rightSeq;
  return String(left?.occurredAt ?? "").localeCompare(String(right?.occurredAt ?? "")) ||
    String(left?.id ?? "").localeCompare(String(right?.id ?? ""));
}

function compareProjectionStateRecords(left, right) {
  return String(left?.updatedAt ?? "").localeCompare(String(right?.updatedAt ?? "")) ||
    String(left?.traceId ?? "").localeCompare(String(right?.traceId ?? ""));
}

function compareWorkbenchFactRecords(left, right) {
  return String(left?.updatedAt ?? left?.createdAt ?? "").localeCompare(String(right?.updatedAt ?? right?.createdAt ?? "")) ||
    String(left?.id ?? left?.sessionId ?? left?.messageId ?? left?.turnId ?? left?.traceId ?? "").localeCompare(String(right?.id ?? right?.sessionId ?? right?.messageId ?? right?.turnId ?? right?.traceId ?? ""));
}

function compareWorkbenchPartFactRecords(left, right) {
  return String(left?.messageId ?? "").localeCompare(String(right?.messageId ?? "")) ||
    nonNegativeInteger(left?.partIndex) - nonNegativeInteger(right?.partIndex) ||
    compareWorkbenchFactRecords(left, right);
}

function compareWorkbenchTraceEventFacts(left, right) {
  return nonNegativeInteger(left?.projectedSeq) - nonNegativeInteger(right?.projectedSeq);
}

function sortWorkbenchFactRows(records, params = {}, family, fallbackCompare) {
  const sorted = [...records];
  if (workbenchFactOrder(params, family) === "updated_desc") {
    return sorted.sort((left, right) => String(right?.updatedAt ?? right?.createdAt ?? "").localeCompare(String(left?.updatedAt ?? left?.createdAt ?? "")) || fallbackCompare(left, right));
  }
  return sorted.sort(fallbackCompare);
}

const WORKBENCH_FACT_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"]);

function workbenchFactFamilySet(params = {}) {
  const raw = params.families ?? params.factFamilies;
  if (raw === undefined || raw === null || raw === "") return new Set(WORKBENCH_FACT_FAMILIES);
  const values = Array.isArray(raw) ? raw : String(raw).split(",");
  const selected = values.map((value) => String(value ?? "").trim()).filter((value) => WORKBENCH_FACT_FAMILIES.includes(value));
  return new Set(selected.length > 0 ? selected : WORKBENCH_FACT_FAMILIES);
}

function workbenchFactFamilyForTable(table) {
  if (table === "workbench_sessions") return "sessions";
  if (table === "workbench_messages") return "messages";
  if (table === "workbench_parts") return "parts";
  if (table === "workbench_turns") return "turns";
  if (table === "workbench_trace_events") return "traceEvents";
  if (table === "workbench_projection_checkpoints") return "checkpoints";
  return "unknown";
}

function workbenchSessionSummarySelectClause() {
  return [
    "session_id",
    "owner_user_id",
    "project_id",
    "conversation_id",
    "thread_id",
    "status",
    "last_trace_id",
    "projected_seq",
    "source_seq",
    "source_event_id",
    "terminal",
    "sealed",
    "created_at",
    "updated_at",
    "COALESCE(NULLIF(session_json::jsonb ->> 'providerProfile', ''), NULLIF(session_json::jsonb #>> '{sessionJson,providerProfile}', '')) AS provider_profile"
  ].join(", ");
}

function workbenchSessionSummaryFactFromRow(row = {}) {
  const sessionId = textOr(row.session_id, "");
  if (!sessionId) return null;
  const providerProfile = textOr(row.provider_profile, "");
  return {
    id: sessionId,
    sessionId,
    ownerUserId: textOr(row.owner_user_id, null),
    projectId: textOr(row.project_id, null),
    conversationId: textOr(row.conversation_id, null),
    threadId: textOr(row.thread_id, null),
    status: textOr(row.status, "unknown"),
    lastTraceId: textOr(row.last_trace_id, null),
    projectedSeq: nonNegativeInteger(row.projected_seq),
    sourceSeq: nonNegativeInteger(row.source_seq),
    sourceEventId: textOr(row.source_event_id, null),
    terminal: row.terminal === true || row.terminal === "t",
    sealed: row.sealed === true || row.sealed === "t",
    ...(providerProfile ? {
      providerProfile,
      sessionJson: {
        providerProfile,
        valuesRedacted: true,
        secretMaterialStored: false
      }
    } : {}),
    createdAt: textOr(row.created_at, null),
    updatedAt: textOr(row.updated_at, row.created_at ?? null),
    valuesPrinted: false,
    valuesRedacted: true
  };
}

function workbenchFactOrder(params = {}, family = "") {
  const value = params[`${family}Order`] ?? params.order;
  return value === "updated_desc" ? "updated_desc" : "updated_asc";
}

function traceEventLevel(value) {
  const text = textOr(value, "info").toLowerCase();
  if (["failed", "failure", "error", "timeout"].includes(text)) return "error";
  if (["warn", "warning", "blocked", "canceled", "cancelled"].includes(text)) return "warn";
  if (["debug", "trace"].includes(text)) return "debug";
  return "info";
}

function positiveIntegerOrNull(value) {
  const parsed = Number.parseInt(value ?? "", 10);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
}

function nonNegativeInteger(value) {
  const parsed = Number.parseInt(value ?? "", 10);
  return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0;
}

function nonNegativeIntegerOrNull(value) {
  const parsed = Number.parseInt(value ?? "", 10);
  return Number.isInteger(parsed) && parsed >= 0 ? parsed : null;
}

function projectionStatusOr(value, fallback) {
  const text = textOr(value, fallback).toLowerCase();
  return ["projecting", "caught_up", "terminal", "degraded", "blocked", "stalled"].includes(text) ? text : fallback;
}

function projectionHealthOr(value, fallback) {
  const text = textOr(value, fallback).toLowerCase();
  return ["healthy", "degraded", "unavailable", "stalled"].includes(text) ? text : fallback;
}

function resultSyncStateOr(value, fallback) {
  const text = textOr(value, fallback).toLowerCase();
  return ["not_started", "pending", "synced", "failed", "timed_out"].includes(text) ? text : fallback;
}

function timestampOr(value, fallback) {
  const text = textOr(value, "");
  if (text) {
    const ms = Date.parse(text);
    if (Number.isFinite(ms)) return new Date(ms).toISOString();
  }
  return textOr(fallback, "") || new Date().toISOString();
}

function textOr(value, fallback) {
  const text = String(value ?? "").trim();
  return text || fallback;
}

function textList(value) {
  const raw = Array.isArray(value) ? value : [];
  return [...new Set(raw.map((item) => textOr(item, "")).filter(Boolean))];
}

function asObject(value, label) {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    throw new HwlabProtocolError(`${label} must be a JSON object`, {
      code: ERROR_CODES.invalidParams
    });
  }
  return value;
}

function requireString(input, field) {
  if (typeof input[field] !== "string" || input[field].trim() === "") {
    throw new HwlabProtocolError(`${field} is required`, {
      code: ERROR_CODES.invalidParams,
      data: { field }
    });
  }
  return input[field];
}

function requireProtocolId(input, field) {
  const value = requireString(input, field);
  return value;
}

function normalizeJsonObject(value) {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    return {};
  }
  return { ...value };
}

function pruneUndefined(input) {
  return Object.fromEntries(Object.entries(input).filter(([, value]) => value !== undefined));
}

function makeId(prefix) {
  return `${prefix}_${randomUUID()}`;
}

function stableJson(value) {
  return JSON.stringify(sortJson(value));
}

function sha256Hex(value) {
  return createHash("sha256").update(stableJson(value)).digest("hex");
}

function sortJson(value) {
  if (Array.isArray(value)) {
    return value.map(sortJson);
  }
  if (value && typeof value === "object") {
    return Object.fromEntries(Object.keys(value).sort().map((key) => [key, sortJson(value[key])]));
  }
  return value;
}

function matchesQuery(record, params, fields) {
  for (const field of fields) {
    if (params[field] !== undefined && params[field] !== null && record[field] !== params[field]) {
      return false;
    }
    const values = textList(params[`${field}s`]);
    if (values.length > 0 && !values.includes(textOr(record[field], ""))) {
      return false;
    }
  }
  return true;
}

function matchesWorkbenchSessionFactQuery(record, params) {
  if (!matchesQuery(record, params, ["sessionId", "ownerUserId", "projectId", "conversationId", "threadId", "status"])) return false;
  const traceId = textOr(params.traceId ?? params.lastTraceId, "");
  if (traceId && record.lastTraceId !== traceId && record.traceId !== traceId) return false;
  return true;
}

function limitResults(records, limit) {
  const parsed = Number.parseInt(limit ?? "", 10);
  if (!Number.isInteger(parsed) || parsed <= 0) {
    return records;
  }
  return records.slice(0, parsed);
}

function normalizeRuntimeAdapter(value) {
  if (typeof value !== "string" || value.trim() === "") {
    return RUNTIME_STORE_KIND;
  }
  const normalized = value.trim().toLowerCase();
  if (["postgres", "postgresql", "pg", "db"].includes(normalized)) {
    return RUNTIME_STORE_KIND_POSTGRES;
  }
  return RUNTIME_STORE_KIND;
}

function normalizePostgresSslMode(value) {
  const normalized = String(value ?? "").trim().toLowerCase();
  if (["disable", "false", "0", "off", "no"].includes(normalized)) {
    return "disable";
  }
  return "require";
}

function postgresSslOption(sslMode) {
  return sslMode === "disable" ? false : { rejectUnauthorized: false };
}

function normalizePostgresConnectionStringForSslMode(dbUrl, sslMode) {
  if (typeof dbUrl !== "string" || dbUrl.trim() === "") {
    return dbUrl;
  }

  let url;
  try {
    url = new URL(dbUrl);
  } catch {
    return dbUrl;
  }

  if (!["postgres:", "postgresql:"].includes(url.protocol)) {
    return dbUrl;
  }

  for (const key of ["ssl", "sslmode", "sslcert", "sslkey", "sslrootcert"]) {
    url.searchParams.delete(key);
  }
  if (sslMode !== "disable") {
    url.searchParams.set("sslmode", "require");
    url.searchParams.set("uselibpqcompat", "true");
  }
  return url.toString();
}

function postgresAdapterContract() {
  return {
    required: "Postgres-backed durable runtime adapter using HWLAB_CLOUD_DB_URL",
    adapterEnv: RUNTIME_ADAPTER_ENV,
    adapterEnvValue: RUNTIME_STORE_KIND_POSTGRES,
    sourceIssue: "pikasTech/HWLAB#164",
    secretMaterialRequiredInSource: false,
    fixtureEvidenceAllowed: false
  };
}

function addRuntimeDurabilityContract(summary) {
  return {
    ...summary,
    durabilityContract: {
      ready: summary.durable === true && summary.ready === true && summary.liveRuntimeEvidence === true,
      status: summary.durable === true && summary.ready === true && summary.liveRuntimeEvidence === true ? "ready" : "blocked",
      requiredEvidence: RUNTIME_DURABILITY_REQUIRED_EVIDENCE,
      dbLiveEvidenceIsDurabilityEvidence: false,
      liveRuntimeEvidence: Boolean(summary.liveRuntimeEvidence),
      adapterQueryRequired: true,
      blockedLayer: summary.durable === true && summary.ready === true && summary.liveRuntimeEvidence === true
        ? null
        : durabilityBlockedLayer(summary),
      blocker: summary.blocker ?? null,
      secretMaterialRead: false
    }
  };
}

function durabilityBlockedLayer(summary = {}) {
  const blocker = summary.blocker;
  if (
    blocker === RUNTIME_DURABLE_ADAPTER_MISSING ||
    blocker === RUNTIME_DURABLE_ADAPTER_UNCONFIGURED ||
    blocker === RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING
  ) {
    return "adapter";
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED) {
    return "ssl";
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED) {
    return "auth";
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED) {
    return "schema";
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED) {
    return "migration";
  }
  if (summary.connection?.queryAttempted || blocker === RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED) {
    return "durability_query";
  }
  return "adapter";
}

function runtimeSafety() {
  return {
    devOnly: true,
    secretsRead: false,
    secretMaterialRead: false,
    valuesRedacted: true,
    endpointRedacted: true
  };
}

export function classifyRuntimeDbError(error) {
  const code = typeof error?.code === "string" ? error.code : "UNKNOWN";
  if (isRuntimeDbConnectTimeout(error)) {
    return {
      blocker: RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
      reason: "Postgres runtime adapter connection attempt timed out before completing the runtime query",
      retryable: true,
      transient: true,
      retryAfterMs: 5000,
      connection: {
        queryAttempted: true,
        queryResult: "connect_timeout",
        errorCode: code
      }
    };
  }
  if (code === "HWLAB_PG_DRIVER_MISSING") {
    return {
      blocker: RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING,
      reason: "Postgres runtime adapter is selected, but the pg package is not installed in the runtime image",
      connection: {
        queryAttempted: false,
        queryResult: "driver_missing",
        errorCode: code
      }
    };
  }
  if (isRuntimeDbSslError(error)) {
    return {
      blocker: RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED,
      reason: "Postgres runtime adapter reached the DB driver but SSL negotiation or SSL mode compatibility blocked the schema query",
      connection: {
        queryAttempted: true,
        queryResult: "ssl_negotiation_blocked",
        errorCode: code
      }
    };
  }
  if (["28P01", "28000", "42501"].includes(code)) {
    return {
      blocker: RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED,
      reason: "Postgres runtime adapter reached the DB driver but authentication or authorization blocked the schema query",
      connection: {
        queryAttempted: true,
        queryResult: "auth_blocked",
        errorCode: code
      }
    };
  }
  if (["42P01", "42703", "3F000"].includes(code)) {
    return {
      blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
      reason: "Postgres runtime adapter reached the database, but the runtime schema is missing or incompatible",
      connection: {
        queryAttempted: true,
        queryResult: "schema_blocked",
        errorCode: code
      }
    };
  }
  return {
    blocker: RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
    reason: "Postgres runtime adapter could not complete a live runtime schema query",
    connection: {
      queryAttempted: true,
      queryResult: "query_blocked",
      errorCode: code
    }
  };
}

function isRuntimeDbConnectTimeout(error) {
  const message = String(error?.message ?? error ?? "").toLowerCase();
  const pgConnectTimeout = message.includes("timeout exceeded when trying to connect");
  const pgConnectionTimeoutTermination = message.includes("connection terminated due to connection timeout");
  if (!pgConnectTimeout && !pgConnectionTimeoutTermination) return false;
  if (pgConnectionTimeoutTermination) return true;
  const stack = String(error?.stack ?? "").toLowerCase();
  return !stack || stack.includes("pg-pool") || stack.includes("node_modules/pg");
}

function durableRuntimeReadBlockedError(method, readiness = {}) {
  return new HwlabProtocolError("Postgres durable runtime adapter is blocked for runtime evidence queries", {
    code: ERROR_CODES.internalError,
    data: durableRuntimeReadBlockedData(method, readiness),
    context: {
      result: "failed"
    }
  });
}

function durableRuntimeReadBlockedData(method, readiness = {}) {
  const durability = readiness.durabilityContract ?? {};
  const connection = readiness.connection ?? {};
  const blockedLayer = durability.blockedLayer ?? durabilityBlockedLayer(readiness);
  const queryResult = connection.queryResult ?? (blockedLayer === "durability_query" ? "query_blocked" : "not_ready");

  return pruneUndefined({
    method,
    adapter: readiness.adapter ?? RUNTIME_STORE_KIND_POSTGRES,
    status: "blocked",
    blocker: readiness.blocker ?? RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
    durable: false,
    durableRequested: true,
    liveRuntimeEvidence: false,
    queryResult,
    blockedLayer,
    requiredEvidence: durability.requiredEvidence ?? RUNTIME_DURABILITY_REQUIRED_EVIDENCE,
    dbLiveEvidenceIsDurabilityEvidence: false,
    retryable: readiness.retryable === true || connection.queryResult === "connect_timeout",
    transient: readiness.transient === true || connection.queryResult === "connect_timeout",
    retryAfterMs: Number.isFinite(Number(readiness.retryAfterMs)) ? Math.max(0, Math.trunc(Number(readiness.retryAfterMs))) : undefined,
    secretMaterialRead: false,
    valueRedacted: true,
    endpointRedacted: true,
    errorCode: connection.errorCode ?? undefined,
    redaction: {
      valueRedacted: true,
      valuesRedacted: true,
      endpointRedacted: true,
      secretMaterialRead: false
    }
  });
}

function isRuntimeDbSslError(error) {
  const code = typeof error?.code === "string" ? error.code : "";
  if (code.startsWith("ERR_SSL") || code.startsWith("ERR_TLS")) {
    return true;
  }
  if ([
    "DEPTH_ZERO_SELF_SIGNED_CERT",
    "SELF_SIGNED_CERT_IN_CHAIN",
    "UNABLE_TO_VERIFY_LEAF_SIGNATURE",
    "UNABLE_TO_GET_ISSUER_CERT_LOCALLY",
    "CERT_HAS_EXPIRED"
  ].includes(code)) {
    return true;
  }
  const message = typeof error?.message === "string" ? error.message.toLowerCase() : "";
  return [
    "does not support ssl",
    "ssl is not enabled",
    "ssl negotiation",
    "ssl off",
    "ssl required",
    "requires ssl",
    "no ssl encryption",
    "no encryption",
    "hostssl",
    "server requires ssl",
    "before secure tls connection",
    "tls handshake"
  ].some((pattern) => message.includes(pattern));
}

function summarizeRuntimeSchema(rows) {
  const columnsByTable = new Map();
  for (const row of rows) {
    const table = row.table_name;
    const column = row.column_name;
    if (!table || !column) continue;
    if (!columnsByTable.has(table)) {
      columnsByTable.set(table, new Set());
    }
    columnsByTable.get(table).add(column);
  }

  const missingTables = [];
  const missingColumns = [];
  for (const [table, columns] of Object.entries(requiredPostgresSchema)) {
    const observed = columnsByTable.get(table);
    if (!observed) {
      missingTables.push(table);
      missingColumns.push(...columns.map((column) => `${table}.${column}`));
      continue;
    }
    for (const column of columns) {
      if (!observed.has(column)) {
        missingColumns.push(`${table}.${column}`);
      }
    }
  }

  return {
    ready: missingTables.length === 0 && missingColumns.length === 0,
    checked: true,
    requiredTables: Object.keys(requiredPostgresSchema),
    missingTables,
    missingColumns
  };
}

function notCheckedRuntimeMigration() {
  return {
    checked: false,
    ready: false,
    table: CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE,
    requiredMigrationId: CLOUD_CORE_MIGRATION_ID,
    requiredSchemaVersion: CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION,
    appliedMigrationId: null,
    appliedSchemaVersion: null,
    missing: true
  };
}

function blockedRuntimeMigration({ errorCode = null } = {}) {
  return {
    ...notCheckedRuntimeMigration(),
    checked: true,
    errorCode
  };
}

function runtimeGates({
  ssl = notCheckedGate(),
  auth = notCheckedGate(),
  schema = notCheckedGate(),
  migration = notCheckedGate(),
  durability = notCheckedGate()
} = {}) {
  return {
    ssl,
    auth,
    schema,
    migration,
    durability
  };
}

function defaultRuntimeGates({ blocker, connection, schema, migration } = {}) {
  if (blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED) {
    return {
      ssl: blockedGate({ checked: true, blocker }),
      auth: notCheckedGate()
    };
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED) {
    return {
      ssl: readyGate(),
      auth: blockedGate({ checked: true, blocker })
    };
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED) {
    return {
      ssl: connection?.queryAttempted ? readyGate() : notCheckedGate(),
      auth: connection?.queryAttempted ? readyGate() : notCheckedGate()
    };
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED && schema?.checked && migration?.checked) {
    return {
      ssl: readyGate(),
      auth: readyGate()
    };
  }
  return {
    ssl: notCheckedGate(),
    auth: notCheckedGate()
  };
}

function defaultSchemaGate({ blocker, schema } = {}) {
  if (schema?.checked) {
    return gateFromReadiness(schema.ready, { blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED });
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED) {
    return blockedGate({ checked: true, blocker });
  }
  return notCheckedGate();
}

function defaultMigrationGate({ blocker, migration } = {}) {
  if (migration?.checked) {
    return gateFromReadiness(migration.ready, { blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED });
  }
  if (blocker === RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED) {
    return blockedGate({ checked: true, blocker });
  }
  return notCheckedGate();
}

function readyGate() {
  return {
    checked: true,
    ready: true,
    status: "ready",
    blocker: null
  };
}

function blockedGate({ checked = true, blocker = "runtime_durable_adapter_blocked" } = {}) {
  return {
    checked,
    ready: false,
    status: checked ? "blocked" : "not_checked",
    blocker
  };
}

function notCheckedGate() {
  return {
    checked: false,
    ready: false,
    status: "not_checked",
    blocker: null
  };
}

function gateFromReadiness(ready, { blocker } = {}) {
  return ready ? readyGate() : blockedGate({ checked: true, blocker });
}

function snapshotMemory(memory) {
  return {
    gatewaySessions: new Map(memory.gatewaySessions),
    boxResources: new Map(memory.boxResources),
    boxCapabilities: new Map(memory.boxCapabilities),
    hardwareOperations: new Map(memory.hardwareOperations),
    auditEvents: new Map(memory.auditEvents),
    evidenceRecords: new Map(memory.evidenceRecords),
    agentTraceEvents: new Map(memory.agentTraceEvents),
    workbenchProjectionStates: new Map(memory.workbenchProjectionStates),
    workbenchSessions: new Map(memory.workbenchSessions),
    workbenchMessages: new Map(memory.workbenchMessages),
    workbenchParts: new Map(memory.workbenchParts),
    workbenchTurns: new Map(memory.workbenchTurns),
    workbenchTraceEvents: new Map(memory.workbenchTraceEvents),
    workbenchProjectionCheckpoints: new Map(memory.workbenchProjectionCheckpoints)
  };
}

function newRecords(current, before) {
  return [...current.entries()]
    .filter(([id]) => !before.has(id))
    .map(([, record]) => record);
}

function changedRecords(current, before) {
  return [...current.entries()]
    .filter(([id, record]) => stableJson(before.get(id)) !== stableJson(record))
    .map(([, record]) => record);
}

function withPersistence(result, persistence) {
  return {
    ...result,
    persistence
  };
}

function parseJsonColumn(value, fallback) {
  if (!value) return fallback;
  if (typeof value === "object") return value;
  try {
    return JSON.parse(String(value));
  } catch {
    return fallback;
  }
}

function toCountKey(table) {
  if (table === "users") return "users";
  if (table === "user_sessions") return "userSessions";
  if (table === "gateway_sessions") return "gatewaySessions";
  if (table === "box_resources") return "boxResources";
  if (table === "box_capabilities") return "boxCapabilities";
  if (table === "hardware_operations") return "hardwareOperations";
  if (table === "audit_events") return "auditEvents";
  if (table === "evidence_records") return "evidenceRecords";
  if (table === "agent_sessions") return "agentSessions";
  if (table === "account_workspaces") return "accountWorkspaces";
  if (table === "worker_sessions") return "workerSessions";
  if (table === "agent_trace_events") return "agentTraceEvents";
  if (table === "workbench_projection_state") return "workbenchProjectionStates";
  if (table === "workbench_sessions") return "workbenchSessions";
  if (table === "workbench_messages") return "workbenchMessages";
  if (table === "workbench_parts") return "workbenchParts";
  if (table === "workbench_turns") return "workbenchTurns";
  if (table === "workbench_trace_events") return "workbenchTraceEvents";
  if (table === "workbench_projection_checkpoints") return "workbenchProjectionCheckpoints";
  return table;
}

function normalizeRuntimeTimeoutMs(value) {
  const parsed = Number.parseInt(String(value ?? ""), 10);
  if (!Number.isInteger(parsed) || parsed <= 0) {
    return 1200;
  }
  return Math.min(parsed, 5000);
}

function normalizeRuntimePoolMax(value) {
  if (value === undefined || value === null || String(value).trim() === "") return null;
  const parsed = Number.parseInt(String(value), 10);
  if (!Number.isInteger(parsed) || parsed <= 0 || String(parsed) !== String(value).trim()) {
    const error = new Error(`${RUNTIME_DB_POOL_MAX_ENV} must be a positive integer`);
    error.code = "HWLAB_CLOUD_DB_POOL_MAX_INVALID";
    throw error;
  }
  return parsed;
}
