lyon
2747 lines
110 KiB
TypeScript
2747 lines
110 KiB
TypeScript
/*
|
|
* 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.
|
|
* 职责: 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 } = {}) {
|
|
const normalizedSslMode = normalizePostgresSslMode(sslMode);
|
|
const normalizedPoolMax = normalizeRuntimePoolMax(poolMax);
|
|
return {
|
|
connectionString: normalizePostgresConnectionStringForSslMode(dbUrl, normalizedSslMode),
|
|
ssl: postgresSslOption(normalizedSslMode),
|
|
connectionTimeoutMillis: normalizeRuntimeTimeoutMs(timeoutMs),
|
|
...(normalizedPoolMax !== null ? { max: normalizedPoolMax } : {})
|
|
};
|
|
}
|
|
|
|
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()
|
|
};
|
|
}
|
|
|
|
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) 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
|
|
} = {}) {
|
|
this.env = env;
|
|
this.dbUrl = dbUrl;
|
|
this.sslMode = sslMode;
|
|
this.now = now;
|
|
this.queryClient = queryClient;
|
|
this.pgModuleLoader = pgModuleLoader;
|
|
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 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);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async queryWorkbenchFacts(params = {}) {
|
|
await this.assertReadyForDurableReads("workbench.facts.query");
|
|
const families = workbenchFactFamilySet(params);
|
|
const entries = await Promise.all([
|
|
["sessions", families.has("sessions") ? 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" }) : Promise.resolve([])],
|
|
["messages", families.has("messages") ? 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" }) : Promise.resolve([])],
|
|
["parts", families.has("parts") ? 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" }) : Promise.resolve([])],
|
|
["turns", families.has("turns") ? this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : Promise.resolve([])],
|
|
["traceEvents", families.has("traceEvents") ? 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" }) : Promise.resolve([])],
|
|
["checkpoints", families.has("checkpoints") ? 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" }) : Promise.resolve([])]
|
|
].map(async ([family, rowsPromise]) => [family, await rowsPromise]));
|
|
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";
|
|
let sql = `SELECT ${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);
|
|
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 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 = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, run_id = EXCLUDED.run_id, command_id = EXCLUDED.command_id, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projection_status = EXCLUDED.projection_status, projection_health = EXCLUDED.projection_health, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, diagnostic_json = EXCLUDED.diagnostic_json, checkpoint_json = EXCLUDED.checkpoint_json, updated_at = EXCLUDED.updated_at",
|
|
[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;
|
|
}
|
|
}
|
|
|
|
async query(sql, params = []) {
|
|
const client = await this.getQueryClient();
|
|
return client.query(sql, params);
|
|
}
|
|
|
|
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,
|
|
poolMax: this.env?.[RUNTIME_DB_POOL_MAX_ENV]
|
|
}));
|
|
return this.pool;
|
|
}
|
|
|
|
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 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 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
|
|
};
|
|
}
|
|
|
|
function classifyRuntimeDbError(error) {
|
|
const code = typeof error?.code === "string" ? error.code : "UNKNOWN";
|
|
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 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,
|
|
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;
|
|
}
|