root
f416950556
Move fresh and retry runner dispatch into AgentRun command admission, remove process-local job creation and secret persistence, and split oversized modules by responsibility. Co-Authored-By: Codex <noreply@openai.com>
1538 lines
74 KiB
TypeScript
1538 lines
74 KiB
TypeScript
/*
|
|
* PostgreSQL Cloud runtime store implementation.
|
|
* Public construction remains in runtime-store.ts.
|
|
*/
|
|
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";
|
|
|
|
|
|
import { CloudRuntimeStore } from "./runtime-store-memory.ts";
|
|
import * as runtimeCore from "./runtime-store-core.ts";
|
|
|
|
const {
|
|
RUNTIME_STORE_KIND,
|
|
RUNTIME_STORE_KIND_POSTGRES,
|
|
RUNTIME_DURABLE_ADAPTER_MISSING,
|
|
RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
|
RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING,
|
|
RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED,
|
|
RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED,
|
|
RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
|
RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED,
|
|
RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
|
|
RUNTIME_DURABILITY_REQUIRED_EVIDENCE,
|
|
RUNTIME_ADAPTER_ENV,
|
|
RUNTIME_DURABLE_ENV,
|
|
RUNTIME_DB_POOL_MAX_ENV,
|
|
requiredPostgresSchema,
|
|
postgresCountTables,
|
|
postgresRuntimeReadIndexes,
|
|
buildPostgresPoolConfig,
|
|
RUNTIME_DB_QUERY_TIMEOUT_MS_ENV,
|
|
RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS_ENV,
|
|
RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS_ENV,
|
|
RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS_ENV,
|
|
RUNTIME_DB_POOL_RESET_COOLDOWN_MS_ENV,
|
|
DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS,
|
|
DEFAULT_RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS,
|
|
DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS,
|
|
DEFAULT_RUNTIME_DB_POOL_RESET_COOLDOWN_MS,
|
|
normalizePositiveInteger,
|
|
delayRuntimeDbQueryRetry,
|
|
isReadyForDurableRuntimeRead,
|
|
normalizeCapabilityInputs,
|
|
normalizeShellInput,
|
|
normalizeAgentTraceEvent,
|
|
normalizeWorkbenchProjectionState,
|
|
normalizeWorkbenchProjectionAllocation,
|
|
normalizeWorkbenchFacts,
|
|
normalizeWorkbenchAggregateEventsForFacts,
|
|
appendWorkbenchFactEvent,
|
|
workbenchAggregateForFact,
|
|
normalizeWorkbenchAggregateEvent,
|
|
workbenchAggregateEventType,
|
|
workbenchInputDelivery,
|
|
workbenchInputStatus,
|
|
indexWorkbenchAggregateEvents,
|
|
normalizeWorkbenchSessionInputFact,
|
|
memoryWorkbenchSessionInputWithSeq,
|
|
inputFactWithAggregateSeq,
|
|
workbenchSessionInputWithSeq,
|
|
workbenchSessionInputPromotedSeq,
|
|
normalizeWorkbenchSessionFact,
|
|
normalizeWorkbenchMessageFact,
|
|
normalizeWorkbenchPartFact,
|
|
normalizeWorkbenchTurnFact,
|
|
assertWorkbenchTurnFinalResponseInvariant,
|
|
normalizeWorkbenchFactStatus,
|
|
workbenchFactFinalResponseText,
|
|
normalizeWorkbenchTraceEventFact,
|
|
normalizeWorkbenchProjectionCheckpoint,
|
|
arrayInput,
|
|
stableWorkbenchFactId,
|
|
requiredWorkbenchFactError,
|
|
compareTraceEventRecords,
|
|
compareProjectionStateRecords,
|
|
compareWorkbenchFactRecords,
|
|
compareWorkbenchInputFactRecords,
|
|
compareWorkbenchPartFactRecords,
|
|
compareWorkbenchTraceEventFacts,
|
|
sortWorkbenchFactRows,
|
|
WORKBENCH_FACT_FAMILIES,
|
|
workbenchFactFamilySet,
|
|
workbenchFactFamilyForTable,
|
|
workbenchSessionSummarySelectClause,
|
|
workbenchSessionSummaryFactFromRow,
|
|
workbenchFactOrder,
|
|
traceEventLevel,
|
|
positiveIntegerOrNull,
|
|
nonNegativeInteger,
|
|
nonNegativeIntegerOrNull,
|
|
projectionStatusOr,
|
|
projectionHealthOr,
|
|
resultSyncStateOr,
|
|
timestampOr,
|
|
textOr,
|
|
textList,
|
|
asObject,
|
|
requireString,
|
|
requireProtocolId,
|
|
normalizeJsonObject,
|
|
pruneUndefined,
|
|
makeId,
|
|
stableJson,
|
|
sha256Hex,
|
|
sortJson,
|
|
matchesQuery,
|
|
matchesWorkbenchSessionFactQuery,
|
|
limitResults,
|
|
normalizeRuntimeAdapter,
|
|
normalizePostgresSslMode,
|
|
postgresSslModeFromConnectionString,
|
|
postgresSslOption,
|
|
normalizePostgresConnectionStringForSslMode,
|
|
postgresAdapterContract,
|
|
addRuntimeDurabilityContract,
|
|
durabilityBlockedLayer,
|
|
runtimeSafety,
|
|
classifyRuntimeDbError,
|
|
isRuntimeDbConnectTimeout,
|
|
durableRuntimeReadBlockedError,
|
|
durableRuntimeReadBlockedData,
|
|
isRuntimeDbSslError,
|
|
summarizeRuntimeSchema,
|
|
notCheckedRuntimeMigration,
|
|
blockedRuntimeMigration,
|
|
runtimeGates,
|
|
defaultRuntimeGates,
|
|
defaultSchemaGate,
|
|
defaultMigrationGate,
|
|
readyGate,
|
|
blockedGate,
|
|
notCheckedGate,
|
|
gateFromReadiness,
|
|
snapshotMemory,
|
|
newRecords,
|
|
changedRecords,
|
|
withPersistence,
|
|
parseJsonColumn,
|
|
postgresPoolStats,
|
|
toCountKey,
|
|
normalizeRuntimeTimeoutMs,
|
|
normalizeRuntimePoolMax
|
|
} = runtimeCore;
|
|
|
|
export class PostgresCloudRuntimeStore {
|
|
constructor({
|
|
env = process.env,
|
|
dbUrl,
|
|
sslMode = "require",
|
|
now = () => new Date().toISOString(),
|
|
queryClient,
|
|
pgModuleLoader,
|
|
logger = console
|
|
} = {}) {
|
|
this.env = env;
|
|
this.dbUrl = dbUrl;
|
|
this.sslMode = sslMode;
|
|
this.now = now;
|
|
this.queryClient = queryClient;
|
|
this.pgModuleLoader = pgModuleLoader;
|
|
this.logger = logger;
|
|
this.pool = null;
|
|
this.postgresPoolResetInFlight = null;
|
|
this.postgresPoolResetLastAtMs = 0;
|
|
this.runtimeReadIndexesReady = false;
|
|
this.runtimeReadIndexesReadyPromise = null;
|
|
this.memory = new CloudRuntimeStore({ now });
|
|
this.lastReadiness = this.blockedSummary({
|
|
blocker: dbUrl || queryClient ? RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED : RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
|
reason: dbUrl || queryClient
|
|
? "Postgres runtime adapter is configured but has not completed a live schema query"
|
|
: "Postgres runtime adapter requires HWLAB_CLOUD_DB_URL or an injected query client"
|
|
});
|
|
}
|
|
|
|
summary() {
|
|
return {
|
|
...this.lastReadiness,
|
|
counts: this.lastReadiness.counts ?? this.memory.summary().counts
|
|
};
|
|
}
|
|
|
|
async readiness() {
|
|
if (!this.dbUrl && !this.queryClient) {
|
|
this.lastReadiness = this.blockedSummary({
|
|
blocker: RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
|
reason: "Postgres runtime adapter is selected but HWLAB_CLOUD_DB_URL is not injected"
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
let rows;
|
|
try {
|
|
const result = await this.query(
|
|
"SELECT table_name, column_name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ANY($1::text[])",
|
|
[Object.keys(requiredPostgresSchema)]
|
|
);
|
|
rows = Array.isArray(result?.rows) ? result.rows : [];
|
|
} catch (error) {
|
|
this.lastReadiness = this.blockedSummary(classifyRuntimeDbError(error));
|
|
return this.summary();
|
|
}
|
|
|
|
const schema = summarizeRuntimeSchema(rows);
|
|
if (!schema.ready) {
|
|
this.lastReadiness = this.blockedSummary({
|
|
blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
|
reason: "Postgres runtime adapter connected, but required runtime tables or columns are missing",
|
|
schema,
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: "schema_blocked"
|
|
},
|
|
gates: runtimeGates({
|
|
ssl: readyGate(),
|
|
auth: readyGate(),
|
|
schema: blockedGate({ checked: true, blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED }),
|
|
migration: notCheckedGate(),
|
|
durability: blockedGate({ checked: false, blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED })
|
|
})
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
let migration;
|
|
try {
|
|
migration = await this.readMigrationReadiness();
|
|
} catch (error) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
const migrationReadBlocker = [
|
|
RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED,
|
|
RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
|
|
].includes(classified.blocker)
|
|
? classified.blocker
|
|
: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED;
|
|
this.lastReadiness = this.blockedSummary({
|
|
blocker: migrationReadBlocker,
|
|
reason: migrationReadBlocker === classified.blocker
|
|
? classified.reason
|
|
: "Postgres runtime schema is present, but the durable runtime migration ledger could not be proven",
|
|
schema,
|
|
migration: blockedRuntimeMigration({ errorCode: classified.connection?.errorCode }),
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: migrationReadBlocker === classified.blocker
|
|
? classified.connection?.queryResult
|
|
: "migration_blocked",
|
|
errorCode: classified.connection?.errorCode ?? null
|
|
},
|
|
gates: runtimeGates({
|
|
ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: readyGate(),
|
|
auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? notCheckedGate()
|
|
: readyGate(),
|
|
schema: readyGate(),
|
|
migration: blockedGate({
|
|
checked: true,
|
|
blocker: migrationReadBlocker
|
|
}),
|
|
durability: blockedGate({ checked: false, blocker: migrationReadBlocker })
|
|
})
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
if (!migration.ready) {
|
|
this.lastReadiness = this.blockedSummary({
|
|
blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED,
|
|
reason: "Postgres runtime schema is present, but the required source migration is not recorded",
|
|
schema,
|
|
migration,
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: "migration_blocked"
|
|
},
|
|
gates: runtimeGates({
|
|
ssl: readyGate(),
|
|
auth: readyGate(),
|
|
schema: readyGate(),
|
|
migration: blockedGate({ checked: true, blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED }),
|
|
durability: blockedGate({ checked: false, blocker: RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED })
|
|
})
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
try {
|
|
await this.ensureRuntimeReadIndexes();
|
|
} catch (error) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
this.lastReadiness = this.blockedSummary({
|
|
...classified,
|
|
schema,
|
|
migration,
|
|
reason: "Postgres runtime schema is present, but Workbench trace read indexes could not be ensured",
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: classified.connection?.queryResult ?? "index_blocked",
|
|
errorCode: classified.connection?.errorCode ?? null
|
|
},
|
|
gates: runtimeGates({
|
|
ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: readyGate(),
|
|
auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? notCheckedGate()
|
|
: readyGate(),
|
|
schema: readyGate(),
|
|
migration: readyGate(),
|
|
durability: blockedGate({ checked: true, blocker: classified.blocker })
|
|
})
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
let counts;
|
|
try {
|
|
counts = await this.readCounts();
|
|
} catch (error) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
this.lastReadiness = this.blockedSummary({
|
|
...classified,
|
|
schema,
|
|
migration,
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? null
|
|
},
|
|
gates: runtimeGates({
|
|
ssl: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: readyGate(),
|
|
auth: classified.blocker === RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED
|
|
? blockedGate({ checked: true, blocker: classified.blocker })
|
|
: classified.blocker === RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED
|
|
? notCheckedGate()
|
|
: readyGate(),
|
|
schema: readyGate(),
|
|
migration: readyGate(),
|
|
durability: blockedGate({ checked: true, blocker: classified.blocker })
|
|
})
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
this.lastReadiness = addRuntimeDurabilityContract({
|
|
adapter: RUNTIME_STORE_KIND_POSTGRES,
|
|
durable: true,
|
|
durableRequested: true,
|
|
durableCapable: true,
|
|
ready: true,
|
|
status: "ready",
|
|
blocker: null,
|
|
reason: "Postgres durable runtime adapter completed live schema, migration, and read readiness queries",
|
|
liveRuntimeEvidence: true,
|
|
fixtureEvidence: false,
|
|
connection: {
|
|
queryAttempted: true,
|
|
queryResult: "durable_readiness_ready",
|
|
endpointRedacted: true,
|
|
valueRedacted: true
|
|
},
|
|
schema,
|
|
migration,
|
|
gates: runtimeGates({
|
|
ssl: readyGate(),
|
|
auth: readyGate(),
|
|
schema: readyGate(),
|
|
migration: readyGate(),
|
|
durability: readyGate()
|
|
}),
|
|
safety: runtimeSafety(),
|
|
adapterContract: postgresAdapterContract(),
|
|
counts
|
|
});
|
|
return this.summary();
|
|
}
|
|
|
|
async registerGatewaySession(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.registerGatewaySession(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async registerBoxResource(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
const input = asObject(params.resource ?? params.boxResource ?? params, "boxResource");
|
|
await this.hydrateGatewaySession(input.gatewaySessionId);
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.registerBoxResource(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async reportBoxCapabilities(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
const inputs = normalizeCapabilityInputs(params);
|
|
for (const input of inputs) {
|
|
await this.hydrateBoxResource(input.resourceId);
|
|
}
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.reportBoxCapabilities(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async requestHardwareOperation(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
await this.hydrateOperationRefs(params);
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.requestHardwareOperation(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async invokeHardwareShell(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
await this.hydrateOperationRefs(params);
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.invokeHardwareShell(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async writeAuditEvent(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.writeAuditEvent(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async queryAuditEvents(params = {}) {
|
|
await this.assertReadyForDurableReads("audit.event.query");
|
|
const result = await this.queryDurableReadRows(
|
|
"audit.event.query",
|
|
"SELECT event_json FROM audit_events ORDER BY timestamp ASC",
|
|
[]
|
|
);
|
|
const events = result.rows
|
|
.map((row) => parseJsonColumn(row.event_json, null))
|
|
.filter(Boolean)
|
|
.filter((event) => matchesQuery(event, params, [
|
|
"auditId",
|
|
"traceId",
|
|
"projectId",
|
|
"gatewaySessionId",
|
|
"operationId",
|
|
"action",
|
|
"targetId"
|
|
]));
|
|
|
|
return {
|
|
events: limitResults(events, params.limit),
|
|
count: events.length,
|
|
persistence: this.summary()
|
|
};
|
|
}
|
|
|
|
async writeEvidenceRecord(params = {}, requestMeta = {}) {
|
|
await this.assertReadyForWrites();
|
|
const before = snapshotMemory(this.memory);
|
|
const result = this.memory.writeEvidenceRecord(params, requestMeta);
|
|
await this.persistChanges(before);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async queryEvidenceRecords(params = {}) {
|
|
await this.assertReadyForDurableReads("evidence.record.query");
|
|
const result = await this.queryDurableReadRows(
|
|
"evidence.record.query",
|
|
"SELECT metadata_json FROM evidence_records ORDER BY created_at ASC",
|
|
[]
|
|
);
|
|
const records = result.rows
|
|
.map((row) => parseJsonColumn(row.metadata_json, null))
|
|
.filter(Boolean)
|
|
.filter((record) => matchesQuery(record, params, [
|
|
"evidenceId",
|
|
"projectId",
|
|
"operationId",
|
|
"agentSessionId",
|
|
"workerSessionId",
|
|
"kind",
|
|
"serviceId"
|
|
]));
|
|
|
|
return {
|
|
records: limitResults(records, params.limit),
|
|
count: records.length,
|
|
persistence: this.summary()
|
|
};
|
|
}
|
|
|
|
async writeAgentTraceEvent(params = {}, requestMeta = {}) {
|
|
const readiness = this.summary();
|
|
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
|
await this.assertReadyForWrites();
|
|
}
|
|
const result = this.memory.writeAgentTraceEvent(params, requestMeta);
|
|
await this.persistAgentTraceEvent(result.traceEvent);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async queryAgentTraceEvents(params = {}) {
|
|
await this.assertReadyForDurableReads("agent.trace-events.query");
|
|
const traceId = textOr(params.traceId, "");
|
|
const sessionId = textOr(params.sessionId, "");
|
|
const workerSessionId = textOr(params.workerSessionId, "");
|
|
const level = textOr(params.level, "");
|
|
const clauses = [];
|
|
const queryParams = [];
|
|
if (traceId) {
|
|
queryParams.push(traceId);
|
|
clauses.push(`trace_id = $${queryParams.length}`);
|
|
}
|
|
if (sessionId) {
|
|
queryParams.push(sessionId);
|
|
clauses.push(`agent_session_id = $${queryParams.length}`);
|
|
}
|
|
if (workerSessionId) {
|
|
queryParams.push(workerSessionId);
|
|
clauses.push(`worker_session_id = $${queryParams.length}`);
|
|
}
|
|
if (level) {
|
|
queryParams.push(level);
|
|
clauses.push(`level = $${queryParams.length}`);
|
|
}
|
|
const sql = `SELECT event_json FROM agent_trace_events${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY occurred_at ASC, id ASC`;
|
|
const result = await this.queryDurableReadRows("agent.trace-events.query", sql, queryParams);
|
|
const events = result.rows
|
|
.map((row) => parseJsonColumn(row.event_json, null))
|
|
.filter(Boolean)
|
|
.filter((event) => matchesQuery(event, params, ["traceId", "sessionId", "workerSessionId", "level"]));
|
|
return {
|
|
events: limitResults(events, params.limit),
|
|
count: events.length,
|
|
persistence: this.summary()
|
|
};
|
|
}
|
|
|
|
async writeWorkbenchProjectionState(params = {}, requestMeta = {}) {
|
|
const readiness = this.summary();
|
|
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
|
await this.assertReadyForWrites();
|
|
}
|
|
const result = this.memory.writeWorkbenchProjectionState(params, requestMeta);
|
|
await this.persistWorkbenchProjectionState(result.projectionState);
|
|
return withPersistence(result, this.summary());
|
|
}
|
|
|
|
async getWorkbenchProjectionState(params = {}) {
|
|
const traceId = textOr(params.traceId, "");
|
|
if (!traceId) return withPersistence({ projectionState: null, found: false }, this.summary());
|
|
const result = await this.queryWorkbenchProjectionStates({ traceId, limit: 1 });
|
|
const projectionState = result.states[0] ?? null;
|
|
return withPersistence({ projectionState, found: Boolean(projectionState) }, result.persistence);
|
|
}
|
|
|
|
async queryWorkbenchProjectionStates(params = {}) {
|
|
await this.assertReadyForDurableReads("workbench.projection-state.query");
|
|
const clauses = [];
|
|
const queryParams = [];
|
|
const addTextClause = (field, column) => {
|
|
const value = textOr(params[field], "");
|
|
if (!value) return;
|
|
queryParams.push(value);
|
|
clauses.push(`${column} = $${queryParams.length}`);
|
|
};
|
|
addTextClause("traceId", "trace_id");
|
|
addTextClause("sessionId", "session_id");
|
|
addTextClause("runId", "run_id");
|
|
addTextClause("commandId", "command_id");
|
|
const statuses = Array.isArray(params.projectionStatuses) ? params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean) : [];
|
|
if (statuses.length > 0) {
|
|
queryParams.push(statuses);
|
|
clauses.push(`projection_status = ANY($${queryParams.length})`);
|
|
}
|
|
const dueAt = textOr(params.dueAt, "");
|
|
if (dueAt) {
|
|
queryParams.push(dueAt);
|
|
clauses.push(`(next_retry_at IS NULL OR next_retry_at <= $${queryParams.length})`);
|
|
}
|
|
const limit = positiveIntegerOrNull(params.limit);
|
|
const offset = positiveIntegerOrNull(params.offset);
|
|
let sql = `SELECT projection_json FROM workbench_projection_state${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ASC, trace_id ASC`;
|
|
if (limit) {
|
|
queryParams.push(limit);
|
|
sql += ` LIMIT $${queryParams.length}`;
|
|
}
|
|
if (offset) {
|
|
queryParams.push(offset);
|
|
sql += ` OFFSET $${queryParams.length}`;
|
|
}
|
|
const result = await this.queryDurableReadRows("workbench.projection-state.query", sql, queryParams);
|
|
const states = result.rows.map((row) => parseJsonColumn(row.projection_json, null)).filter(Boolean);
|
|
return {
|
|
states,
|
|
count: states.length,
|
|
persistence: this.summary()
|
|
};
|
|
}
|
|
|
|
async allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) {
|
|
const readiness = this.summary();
|
|
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
|
await this.assertReadyForWrites();
|
|
}
|
|
const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now());
|
|
const result = await this.withDurableTransaction("workbench.projected-seq.allocate", async (client) => {
|
|
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${allocation.traceId}`]);
|
|
const existing = await client.query(
|
|
"SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2 ORDER BY projected_seq ASC LIMIT 1",
|
|
[allocation.traceId, allocation.sourceEventId]
|
|
);
|
|
const existingSeq = nonNegativeInteger(existing.rows?.[0]?.projected_seq);
|
|
if (existingSeq > 0) return { projectedSeq: existingSeq, reused: true };
|
|
const maxResult = await client.query(
|
|
"SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq",
|
|
[allocation.traceId]
|
|
);
|
|
const proposedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1;
|
|
const reserve = await client.query(
|
|
"INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq + 1, EXCLUDED.projected_seq), updated_at = EXCLUDED.updated_at RETURNING projected_seq",
|
|
[allocation.traceId, allocation.sessionId, allocation.turnId, allocation.runId, allocation.commandId, proposedSeq, allocation.sourceSeq, allocation.sourceEventId, "projecting", "healthy", false, false, stableJson({ allocator: "workbench-projected-seq", valuesRedacted: true }), stableJson({ ...allocation, projectedSeq: proposedSeq, projectionStatus: "projecting", projectionHealth: "healthy", valuesPrinted: false }), allocation.createdAt, allocation.updatedAt]
|
|
);
|
|
return { projectedSeq: nonNegativeInteger(reserve.rows?.[0]?.projected_seq) || proposedSeq, reused: false };
|
|
});
|
|
return {
|
|
...allocation,
|
|
projectedSeq: result.projectedSeq,
|
|
reused: result.reused,
|
|
persistence: this.summary(),
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
async writeWorkbenchFacts(params = {}, requestMeta = {}) {
|
|
const readiness = this.summary();
|
|
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
|
await this.assertReadyForWrites();
|
|
}
|
|
const facts = normalizeWorkbenchFacts(params.facts ?? params, requestMeta, this.now());
|
|
const aggregateEvents = normalizeWorkbenchAggregateEventsForFacts(facts, requestMeta, this.now());
|
|
const txResult = await this.withDurableTransaction("workbench.facts.write", async (client) => {
|
|
const persistedEvents = [];
|
|
for (const event of aggregateEvents) persistedEvents.push(await this.persistWorkbenchAggregateEvent(event, client));
|
|
const eventIndex = indexWorkbenchAggregateEvents(persistedEvents);
|
|
const inputFacts = facts.inputs.map((fact) => inputFactWithAggregateSeq(fact, eventIndex));
|
|
for (const fact of inputFacts) await this.persistWorkbenchSessionInputFact(fact, client);
|
|
for (const fact of facts.sessions) await this.persistWorkbenchSessionFact(fact, client);
|
|
for (const fact of facts.messages) await this.persistWorkbenchMessageFact(fact, client);
|
|
for (const fact of facts.parts) await this.persistWorkbenchPartFact(fact, client);
|
|
for (const fact of facts.turns) await this.persistWorkbenchTurnFact(fact, client);
|
|
for (const fact of facts.traceEvents) await this.persistWorkbenchTraceEventFact(fact, client);
|
|
for (const fact of facts.checkpoints) await this.persistWorkbenchProjectionCheckpoint(fact, client);
|
|
for (const fact of facts.messages) {
|
|
const event = eventIndex.get(`message:${fact.messageId}`) ?? eventIndex.get(`messages:${fact.messageId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await this.persistWorkbenchProjectionOutbox({
|
|
eventSeq: event?.eventSeq ?? null,
|
|
aggregateId: event?.aggregateId ?? null,
|
|
aggregateSeq: event?.aggregateSeq ?? 0,
|
|
projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
|
|
traceId: fact.traceId,
|
|
sessionId: fact.sessionId,
|
|
turnId: fact.turnId,
|
|
messageId: fact.messageId,
|
|
projectedSeq: fact.projectedSeq,
|
|
sourceSeq: fact.sourceSeq,
|
|
sourceEventId: fact.sourceEventId,
|
|
commitType: "message",
|
|
terminal: Boolean(fact.terminal),
|
|
sealed: Boolean(fact.sealed),
|
|
payload: { role: fact.role, status: fact.status, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true },
|
|
createdAt: fact.updatedAt ?? fact.createdAt ?? this.now()
|
|
}, client);
|
|
}
|
|
for (const fact of facts.traceEvents) {
|
|
const event = eventIndex.get(`traceEvent:${fact.id}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await this.persistWorkbenchProjectionOutbox({
|
|
eventSeq: event?.eventSeq ?? null,
|
|
aggregateId: event?.aggregateId ?? null,
|
|
aggregateSeq: event?.aggregateSeq ?? 0,
|
|
projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
|
|
traceId: fact.traceId,
|
|
sessionId: fact.sessionId,
|
|
turnId: fact.turnId,
|
|
messageId: fact.messageId,
|
|
projectedSeq: fact.projectedSeq,
|
|
sourceSeq: fact.sourceSeq,
|
|
sourceEventId: fact.sourceEventId,
|
|
commitType: fact.terminal ? "terminal" : "event",
|
|
terminal: Boolean(fact.terminal),
|
|
sealed: Boolean(fact.sealed),
|
|
payload: { type: fact.eventType, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true },
|
|
createdAt: fact.occurredAt ?? fact.updatedAt ?? this.now()
|
|
}, client);
|
|
}
|
|
for (const fact of facts.turns) {
|
|
if (fact.terminal || fact.sealed) {
|
|
const event = eventIndex.get(`turn:${fact.turnId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await this.persistWorkbenchProjectionOutbox({
|
|
eventSeq: event?.eventSeq ?? null,
|
|
aggregateId: event?.aggregateId ?? null,
|
|
aggregateSeq: event?.aggregateSeq ?? 0,
|
|
projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
|
|
traceId: fact.traceId,
|
|
sessionId: fact.sessionId,
|
|
turnId: fact.turnId,
|
|
messageId: fact.messageId,
|
|
projectedSeq: fact.projectedSeq,
|
|
sourceSeq: fact.sourceSeq,
|
|
sourceEventId: fact.sourceEventId,
|
|
commitType: "terminal",
|
|
terminal: true,
|
|
sealed: true,
|
|
payload: { status: fact.status, failureKind: fact.failureKind, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, valuesRedacted: true },
|
|
createdAt: fact.updatedAt ?? this.now()
|
|
}, client);
|
|
}
|
|
}
|
|
return { events: persistedEvents, inputFacts };
|
|
});
|
|
const persistedFacts = { ...facts, inputs: txResult.inputFacts };
|
|
this.memory.writeWorkbenchFacts({ facts: persistedFacts }, requestMeta);
|
|
return withPersistence({ written: true, facts: persistedFacts, events: txResult.events }, this.summary());
|
|
}
|
|
|
|
async queryWorkbenchFacts(params = {}) {
|
|
await this.assertReadyForDurableReads("workbench.facts.query");
|
|
const families = workbenchFactFamilySet(params);
|
|
// A single Workbench read can request several fact families. Running those
|
|
// SQL reads in parallel lets one HTTP request occupy the whole Postgres
|
|
// pool, which makes the control+observer pages amplify into 503s under
|
|
// periodic refresh. Keep this read model bounded to one connection per
|
|
// request; the page-level latency is preferable to pool starvation.
|
|
const entries = [];
|
|
entries.push(["inputs", families.has("inputs") ? await this.queryWorkbenchFactRows("workbench.session-inputs.query", "workbench_session_inputs", "input_json", params, { inputId: "input_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", messageId: "message_id", commandId: "command_id", delivery: "delivery", status: "status" }) : []]);
|
|
entries.push(["sessions", families.has("sessions") ? await this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }) : []]);
|
|
entries.push(["messages", families.has("messages") ? await this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }) : []]);
|
|
entries.push(["parts", families.has("parts") ? await this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }) : []]);
|
|
entries.push(["turns", families.has("turns") ? await this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : []]);
|
|
entries.push(["traceEvents", families.has("traceEvents") ? await this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }) : []]);
|
|
entries.push(["checkpoints", families.has("checkpoints") ? await this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" }) : []]);
|
|
const facts = Object.fromEntries(entries);
|
|
return withPersistence({ facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0) }, this.summary());
|
|
}
|
|
|
|
async queryWorkbenchFactRows(method, table, jsonColumn, params = {}, columnMap = {}) {
|
|
const family = workbenchFactFamilyForTable(table);
|
|
const queryParams = [];
|
|
const clauses = [];
|
|
for (const [field, column] of Object.entries(columnMap)) {
|
|
const value = textOr(params[field], "");
|
|
if (value) {
|
|
queryParams.push(value);
|
|
clauses.push(`${column} = $${queryParams.length}`);
|
|
}
|
|
const values = textList(params[`${field}s`]);
|
|
if (values.length > 0) {
|
|
queryParams.push(values);
|
|
clauses.push(`${column} = ANY($${queryParams.length})`);
|
|
}
|
|
}
|
|
const limit = positiveIntegerOrNull(params.limit);
|
|
if (family === "traceEvents") {
|
|
const afterProjectedSeq = nonNegativeInteger(params.afterProjectedSeq);
|
|
if (afterProjectedSeq > 0) {
|
|
queryParams.push(afterProjectedSeq);
|
|
clauses.push(`projected_seq > $${queryParams.length}`);
|
|
}
|
|
let sql = `SELECT ${jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY projected_seq ASC`;
|
|
if (limit) {
|
|
queryParams.push(limit);
|
|
sql += ` LIMIT $${queryParams.length}`;
|
|
}
|
|
const result = await this.queryDurableReadRows(method, sql, queryParams);
|
|
return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
|
|
}
|
|
const orderDirection = workbenchFactOrder(params, family) === "updated_desc" ? "DESC" : "ASC";
|
|
const useSessionSummaryProjection = table === "workbench_sessions" && params.sessionProjection === "summary";
|
|
const orderClause = family === "inputs" && orderDirection === "ASC" ? "admitted_seq ASC, input_id ASC" : `updated_at ${orderDirection}`;
|
|
let sql = `SELECT ${useSessionSummaryProjection ? workbenchSessionSummarySelectClause() : jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY ${orderClause}`;
|
|
if (limit) {
|
|
queryParams.push(limit);
|
|
sql += ` LIMIT $${queryParams.length}`;
|
|
}
|
|
const result = await this.queryDurableReadRows(method, sql, queryParams);
|
|
if (useSessionSummaryProjection) return result.rows.map(workbenchSessionSummaryFactFromRow).filter(Boolean);
|
|
return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
|
|
}
|
|
|
|
async hydrateOperationRefs(params = {}) {
|
|
await this.hydrateGatewaySession(params.gatewaySessionId);
|
|
await this.hydrateBoxResource(params.resourceId);
|
|
await this.hydrateBoxCapability(params.capabilityId);
|
|
}
|
|
|
|
async hydrateGatewaySession(gatewaySessionId) {
|
|
if (!gatewaySessionId || this.memory.gatewaySessions.has(gatewaySessionId)) return;
|
|
const result = await this.query("SELECT gateway_session_json FROM gateway_sessions WHERE id = $1 LIMIT 1", [
|
|
gatewaySessionId
|
|
]);
|
|
const record = parseJsonColumn(result.rows?.[0]?.gateway_session_json, null);
|
|
if (record) this.memory.gatewaySessions.set(record.gatewaySessionId, record);
|
|
}
|
|
|
|
async hydrateBoxResource(resourceId) {
|
|
if (!resourceId || this.memory.boxResources.has(resourceId)) return;
|
|
const result = await this.query("SELECT resource_json FROM box_resources WHERE id = $1 LIMIT 1", [
|
|
resourceId
|
|
]);
|
|
const record = parseJsonColumn(result.rows?.[0]?.resource_json, null);
|
|
if (record) {
|
|
await this.hydrateGatewaySession(record.gatewaySessionId);
|
|
this.memory.boxResources.set(record.resourceId, record);
|
|
}
|
|
}
|
|
|
|
async hydrateBoxCapability(capabilityId) {
|
|
if (!capabilityId || this.memory.boxCapabilities.has(capabilityId)) return;
|
|
const result = await this.query("SELECT capability_json FROM box_capabilities WHERE id = $1 LIMIT 1", [
|
|
capabilityId
|
|
]);
|
|
const record = parseJsonColumn(result.rows?.[0]?.capability_json, null);
|
|
if (record) {
|
|
await this.hydrateBoxResource(record.resourceId);
|
|
this.memory.boxCapabilities.set(record.capabilityId, record);
|
|
}
|
|
}
|
|
|
|
async persistChanges(before) {
|
|
for (const record of newRecords(this.memory.gatewaySessions, before.gatewaySessions)) {
|
|
await this.persistGatewaySession(record);
|
|
}
|
|
for (const record of newRecords(this.memory.boxResources, before.boxResources)) {
|
|
await this.persistBoxResource(record);
|
|
}
|
|
for (const record of newRecords(this.memory.boxCapabilities, before.boxCapabilities)) {
|
|
await this.persistBoxCapability(record);
|
|
}
|
|
for (const record of changedRecords(this.memory.hardwareOperations, before.hardwareOperations)) {
|
|
await this.persistHardwareOperation(record);
|
|
}
|
|
for (const record of newRecords(this.memory.auditEvents, before.auditEvents)) {
|
|
await this.persistAuditEvent(record);
|
|
}
|
|
for (const record of newRecords(this.memory.evidenceRecords, before.evidenceRecords)) {
|
|
await this.persistEvidenceRecord(record);
|
|
}
|
|
for (const record of newRecords(this.memory.agentTraceEvents, before.agentTraceEvents)) {
|
|
await this.persistAgentTraceEvent(record);
|
|
}
|
|
for (const record of changedRecords(this.memory.workbenchProjectionStates, before.workbenchProjectionStates)) {
|
|
await this.persistWorkbenchProjectionState(record);
|
|
}
|
|
for (const record of changedRecords(this.memory.workbenchSessionInputs, before.workbenchSessionInputs)) {
|
|
await this.persistWorkbenchSessionInputFact(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() {
|
|
if (!this.dbUrl && !this.queryClient) {
|
|
const readiness = this.blockedSummary({
|
|
blocker: RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
|
reason: "Postgres runtime adapter is selected but HWLAB_CLOUD_DB_URL is not injected"
|
|
});
|
|
this.lastReadiness = readiness;
|
|
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 persistWorkbenchAggregateEvent(record, client = this) {
|
|
const existing = await client.query(
|
|
"SELECT event_seq, aggregate_seq, projection_revision FROM workbench_events WHERE event_id = $1 OR ($2::text IS NOT NULL AND aggregate_id = $3 AND source_event_id = $2) ORDER BY event_seq ASC LIMIT 1",
|
|
[record.eventId, record.sourceEventId, record.aggregateId]
|
|
);
|
|
const existingRow = existing.rows?.[0] ?? null;
|
|
if (existingRow) {
|
|
return {
|
|
...record,
|
|
eventSeq: Number(existingRow.event_seq),
|
|
aggregateSeq: Number(existingRow.aggregate_seq),
|
|
projectionRevision: Number(existingRow.projection_revision)
|
|
};
|
|
}
|
|
await client.query(
|
|
"INSERT INTO workbench_event_sequences (aggregate_id, aggregate_type, session_id, turn_id, trace_id, last_seq, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,0,$6,$7) ON CONFLICT (aggregate_id) DO UPDATE SET aggregate_type = EXCLUDED.aggregate_type, session_id = COALESCE(workbench_event_sequences.session_id, EXCLUDED.session_id), turn_id = COALESCE(workbench_event_sequences.turn_id, EXCLUDED.turn_id), trace_id = COALESCE(workbench_event_sequences.trace_id, EXCLUDED.trace_id), updated_at = EXCLUDED.updated_at",
|
|
[record.aggregateId, record.aggregateType, record.sessionId, record.turnId, record.traceId, record.committedAt, record.committedAt]
|
|
);
|
|
const reserved = await client.query(
|
|
"UPDATE workbench_event_sequences SET last_seq = last_seq + 1, updated_at = $2 WHERE aggregate_id = $1 RETURNING last_seq",
|
|
[record.aggregateId, record.committedAt]
|
|
);
|
|
const aggregateSeq = nonNegativeInteger(reserved.rows?.[0]?.last_seq);
|
|
const projectionRevision = nonNegativeInteger(record.projectionRevision) || aggregateSeq;
|
|
const inserted = await client.query(
|
|
"INSERT INTO workbench_events (event_id, aggregate_id, aggregate_type, aggregate_seq, session_id, turn_id, trace_id, message_id, source_run_id, source_command_id, source_seq, source_event_id, event_type, projection_revision, terminal, sealed, payload_json, occurred_at, committed_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19) ON CONFLICT (event_id) DO UPDATE SET projection_revision = GREATEST(workbench_events.projection_revision, EXCLUDED.projection_revision), terminal = workbench_events.terminal OR EXCLUDED.terminal, sealed = workbench_events.sealed OR EXCLUDED.sealed, payload_json = EXCLUDED.payload_json, committed_at = EXCLUDED.committed_at RETURNING event_seq, aggregate_seq, projection_revision",
|
|
[record.eventId, record.aggregateId, record.aggregateType, aggregateSeq, record.sessionId, record.turnId, record.traceId, record.messageId, record.sourceRunId, record.sourceCommandId, record.sourceSeq, record.sourceEventId, record.eventType, projectionRevision, Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.occurredAt, record.committedAt]
|
|
);
|
|
const row = inserted.rows?.[0] ?? {};
|
|
return {
|
|
...record,
|
|
eventSeq: Number(row.event_seq),
|
|
aggregateSeq: Number(row.aggregate_seq) || aggregateSeq,
|
|
projectionRevision: Number(row.projection_revision) || projectionRevision
|
|
};
|
|
}
|
|
|
|
async persistWorkbenchSessionFact(record, client = this) {
|
|
await client.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 persistWorkbenchSessionInputFact(record, client = this) {
|
|
await client.query(
|
|
"INSERT INTO workbench_session_inputs (input_id, session_id, turn_id, trace_id, message_id, command_id, delivery, admitted_seq, promoted_seq, status, error_code, source_event_id, input_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (input_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, message_id = EXCLUDED.message_id, command_id = COALESCE(EXCLUDED.command_id, workbench_session_inputs.command_id), delivery = EXCLUDED.delivery, admitted_seq = CASE WHEN workbench_session_inputs.admitted_seq > 0 THEN workbench_session_inputs.admitted_seq ELSE EXCLUDED.admitted_seq END, promoted_seq = COALESCE(EXCLUDED.promoted_seq, workbench_session_inputs.promoted_seq), status = EXCLUDED.status, error_code = EXCLUDED.error_code, source_event_id = COALESCE(EXCLUDED.source_event_id, workbench_session_inputs.source_event_id), input_json = EXCLUDED.input_json, updated_at = EXCLUDED.updated_at",
|
|
[record.inputId, record.sessionId, record.turnId, record.traceId, record.messageId, record.commandId, record.delivery, record.admittedSeq, record.promotedSeq, record.status, record.errorCode, record.sourceEventId, stableJson(record), record.createdAt, record.updatedAt]
|
|
);
|
|
}
|
|
|
|
async persistWorkbenchMessageFact(record, client = this) {
|
|
await client.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, client = this) {
|
|
await client.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, client = this) {
|
|
await client.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, client = this) {
|
|
await client.query(
|
|
"INSERT INTO workbench_trace_events (id, trace_id, session_id, turn_id, message_id, source_seq, source_event_id, projected_seq, event_type, terminal, sealed, event_json, occurred_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (id) DO UPDATE SET trace_id = EXCLUDED.trace_id, session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, message_id = EXCLUDED.message_id, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projected_seq = EXCLUDED.projected_seq, event_type = EXCLUDED.event_type, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, event_json = EXCLUDED.event_json, occurred_at = EXCLUDED.occurred_at, updated_at = EXCLUDED.updated_at",
|
|
[record.id, record.traceId, record.sessionId, record.turnId, record.messageId, record.sourceSeq, record.sourceEventId, record.projectedSeq, record.eventType, record.terminal, record.sealed, stableJson(record), record.occurredAt, record.updatedAt]
|
|
);
|
|
}
|
|
|
|
async persistWorkbenchProjectionOutbox(record, client = this) {
|
|
await client.query(
|
|
"INSERT INTO workbench_projection_outbox (event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)",
|
|
[record.eventSeq, record.aggregateId, nonNegativeInteger(record.aggregateSeq), nonNegativeInteger(record.projectionRevision), record.traceId, record.sessionId, record.turnId, record.messageId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.commitType ?? "event", Boolean(record.terminal), Boolean(record.sealed), stableJson(record.payload ?? record), record.createdAt ?? this.now()]
|
|
);
|
|
}
|
|
|
|
async readWorkbenchProjectionOutbox({ afterSeq = 0, limit = 100, traceId = null, sessionId = null } = {}) {
|
|
const params = [];
|
|
let where = "outbox_seq > $1";
|
|
params.push(Number(afterSeq) || 0);
|
|
let paramIdx = 2;
|
|
if (traceId) {
|
|
where += ` AND trace_id = $${paramIdx}`;
|
|
params.push(traceId);
|
|
paramIdx += 1;
|
|
}
|
|
if (sessionId) {
|
|
where += ` AND session_id = $${paramIdx}`;
|
|
params.push(sessionId);
|
|
paramIdx += 1;
|
|
}
|
|
params.push(Math.min(Math.max(Number(limit) || 100, 1), 500));
|
|
const result = await this.query(
|
|
`SELECT outbox_seq, event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE ${where} ORDER BY outbox_seq ASC LIMIT $${paramIdx}`,
|
|
params
|
|
);
|
|
return (result.rows ?? []).map((row) => ({
|
|
outboxSeq: Number(row.outbox_seq),
|
|
eventSeq: row.event_seq === null || row.event_seq === undefined ? null : Number(row.event_seq),
|
|
aggregateId: row.aggregate_id,
|
|
aggregateSeq: Number(row.aggregate_seq ?? 0),
|
|
projectionRevision: Number(row.projection_revision ?? 0),
|
|
traceId: row.trace_id,
|
|
sessionId: row.session_id,
|
|
turnId: row.turn_id,
|
|
messageId: row.message_id,
|
|
projectedSeq: Number(row.projected_seq),
|
|
sourceSeq: Number(row.source_seq),
|
|
sourceEventId: row.source_event_id,
|
|
commitType: row.commit_type,
|
|
terminal: Boolean(row.terminal),
|
|
sealed: Boolean(row.sealed),
|
|
payload: typeof row.payload_json === "string" ? JSON.parse(row.payload_json) : row.payload_json,
|
|
createdAt: row.created_at,
|
|
valuesPrinted: false
|
|
}));
|
|
}
|
|
|
|
async persistWorkbenchProjectionCheckpoint(record, client = this) {
|
|
await client.query(
|
|
"INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.session_id ELSE workbench_projection_checkpoints.session_id END, turn_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.turn_id ELSE workbench_projection_checkpoints.turn_id END, run_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.run_id ELSE workbench_projection_checkpoints.run_id END, command_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.command_id ELSE workbench_projection_checkpoints.command_id END, projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq, EXCLUDED.projected_seq), source_seq = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_seq ELSE workbench_projection_checkpoints.source_seq END, source_event_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_event_id ELSE workbench_projection_checkpoints.source_event_id END, projection_status = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_status ELSE workbench_projection_checkpoints.projection_status END, projection_health = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_health ELSE workbench_projection_checkpoints.projection_health END, terminal = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.terminal ELSE workbench_projection_checkpoints.terminal END, sealed = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.sealed ELSE workbench_projection_checkpoints.sealed END, diagnostic_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.diagnostic_json ELSE workbench_projection_checkpoints.diagnostic_json END, checkpoint_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.checkpoint_json ELSE workbench_projection_checkpoints.checkpoint_json END, updated_at = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.updated_at ELSE workbench_projection_checkpoints.updated_at END",
|
|
[record.traceId, record.sessionId, record.turnId, record.runId, record.commandId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.projectionStatus, record.projectionHealth, record.terminal, record.sealed, stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt]
|
|
);
|
|
}
|
|
|
|
async readCounts() {
|
|
const counts = {};
|
|
for (const table of postgresCountTables) {
|
|
const result = await this.query(`SELECT COUNT(*)::int AS count FROM ${table}`, []);
|
|
counts[toCountKey(table)] = Number(result.rows?.[0]?.count ?? 0);
|
|
}
|
|
return counts;
|
|
}
|
|
|
|
async readMigrationReadiness() {
|
|
const result = await this.query(
|
|
`SELECT id, schema_version FROM ${CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE} WHERE id = $1 LIMIT 1`,
|
|
[CLOUD_CORE_MIGRATION_ID]
|
|
);
|
|
const row = result.rows?.[0] ?? null;
|
|
const ready =
|
|
row?.id === CLOUD_CORE_MIGRATION_ID &&
|
|
row?.schema_version === CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION;
|
|
return {
|
|
checked: true,
|
|
ready,
|
|
table: CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE,
|
|
requiredMigrationId: CLOUD_CORE_MIGRATION_ID,
|
|
requiredSchemaVersion: CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION,
|
|
appliedMigrationId: row?.id ?? null,
|
|
appliedSchemaVersion: row?.schema_version ?? null,
|
|
missing: !row
|
|
};
|
|
}
|
|
|
|
async ensureRuntimeReadIndexes() {
|
|
if (this.runtimeReadIndexesReady) return;
|
|
if (!this.runtimeReadIndexesReadyPromise) {
|
|
this.runtimeReadIndexesReadyPromise = (async () => {
|
|
for (const sql of postgresRuntimeReadIndexes) {
|
|
await this.query(sql, []);
|
|
}
|
|
this.runtimeReadIndexesReady = true;
|
|
})();
|
|
}
|
|
try {
|
|
await this.runtimeReadIndexesReadyPromise;
|
|
} finally {
|
|
if (!this.runtimeReadIndexesReady) this.runtimeReadIndexesReadyPromise = null;
|
|
}
|
|
}
|
|
|
|
recordRuntimeDbOperationFailure(label, error, retry = {}) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
this.lastReadiness = this.blockedSummary({
|
|
...classified,
|
|
reason: `Postgres durable runtime adapter could not complete ${label}`
|
|
});
|
|
this.logger?.warn?.({
|
|
event: "postgres_runtime_operation_failed",
|
|
operation: label,
|
|
blocker: classified.blocker,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
|
retryable: classified.retryable === true,
|
|
transient: classified.transient === true,
|
|
retryAfterMs: classified.retryAfterMs ?? null,
|
|
retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
|
|
retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
|
|
retrying: retry.retrying === true,
|
|
retryDelayMs: Number.isFinite(Number(retry.delayMs)) ? Number(retry.delayMs) : null,
|
|
retryLabel: retry.label ?? null,
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
});
|
|
}
|
|
|
|
resetPostgresPoolAfterTransientFailure(classified, retry = {}) {
|
|
if (classified?.connection?.queryResult !== "connect_timeout") return false;
|
|
const pool = this.pool;
|
|
if (!pool) return false;
|
|
const now = Date.now();
|
|
const cooldownMs = this.runtimePoolResetCooldownMs();
|
|
if (this.postgresPoolResetInFlight) {
|
|
this.logger?.warn?.({
|
|
event: "postgres_runtime_pool_reset_skipped",
|
|
reason: "in_flight",
|
|
blocker: classified.blocker,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
|
retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
|
|
retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
|
|
cooldownMs,
|
|
poolStats: postgresPoolStats(pool),
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
});
|
|
return false;
|
|
}
|
|
if (now - this.postgresPoolResetLastAtMs < cooldownMs) {
|
|
this.logger?.warn?.({
|
|
event: "postgres_runtime_pool_reset_skipped",
|
|
reason: "cooldown",
|
|
blocker: classified.blocker,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
|
retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
|
|
retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
|
|
cooldownMs,
|
|
poolResetLastAtMs: this.postgresPoolResetLastAtMs,
|
|
poolStats: postgresPoolStats(pool),
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
});
|
|
return false;
|
|
}
|
|
this.pool = null;
|
|
this.postgresPoolResetLastAtMs = now;
|
|
this.logger?.warn?.({
|
|
event: "postgres_runtime_pool_reset_after_transient_failure",
|
|
blocker: classified.blocker,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
|
retryAttempt: Number.isFinite(Number(retry.attempt)) ? Number(retry.attempt) : null,
|
|
retryMaxAttempts: Number.isFinite(Number(retry.maxAttempts)) ? Number(retry.maxAttempts) : null,
|
|
retrying: retry.retrying === true,
|
|
cooldownMs,
|
|
poolStats: postgresPoolStats(pool),
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
});
|
|
if (typeof pool.end === "function") {
|
|
this.postgresPoolResetInFlight = Promise.resolve()
|
|
.then(() => pool.end())
|
|
.catch((error) => this.logger?.warn?.({
|
|
event: "postgres_runtime_pool_reset_failed",
|
|
errorCode: error?.code ?? "UNKNOWN",
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
}))
|
|
.finally(() => {
|
|
this.postgresPoolResetInFlight = null;
|
|
});
|
|
}
|
|
return true;
|
|
}
|
|
|
|
runtimePoolResetCooldownMs() {
|
|
const env = this.env ?? {};
|
|
return normalizePositiveInteger(env[RUNTIME_DB_POOL_RESET_COOLDOWN_MS_ENV], DEFAULT_RUNTIME_DB_POOL_RESET_COOLDOWN_MS);
|
|
}
|
|
|
|
async query(sql, params = []) {
|
|
const retry = this.runtimeQueryRetryPolicy();
|
|
for (let attempt = 1; attempt <= retry.maxAttempts; attempt += 1) {
|
|
try {
|
|
const client = await this.getQueryClient();
|
|
return await client.query(sql, params);
|
|
} catch (error) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
const retrying = classified.retryable === true && classified.transient === true && attempt < retry.maxAttempts;
|
|
const delayMs = retrying ? Math.min(retry.maxDelayMs, retry.initialDelayMs * (2 ** Math.max(0, attempt - 1))) : null;
|
|
this.recordRuntimeDbOperationFailure("query", error, {
|
|
attempt,
|
|
maxAttempts: retry.maxAttempts,
|
|
retrying,
|
|
delayMs,
|
|
label: `${attempt}/${retry.maxAttempts}`
|
|
});
|
|
this.resetPostgresPoolAfterTransientFailure(classified, {
|
|
attempt,
|
|
maxAttempts: retry.maxAttempts,
|
|
retrying,
|
|
delayMs,
|
|
label: `${attempt}/${retry.maxAttempts}`
|
|
});
|
|
if (!retrying) throw error;
|
|
await delayRuntimeDbQueryRetry(delayMs);
|
|
}
|
|
}
|
|
}
|
|
|
|
runtimeQueryRetryPolicy() {
|
|
const env = this.env ?? {};
|
|
return {
|
|
maxAttempts: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_ATTEMPTS),
|
|
initialDelayMs: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_INITIAL_DELAY_MS),
|
|
maxDelayMs: normalizePositiveInteger(env[RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS_ENV], DEFAULT_RUNTIME_DB_QUERY_RETRY_MAX_DELAY_MS)
|
|
};
|
|
}
|
|
|
|
async withDurableTransaction(label, fn) {
|
|
try {
|
|
const client = await this.getQueryClient();
|
|
if (typeof client.connect === "function") {
|
|
const tx = await client.connect();
|
|
try {
|
|
await tx.query("BEGIN");
|
|
const result = await fn(tx);
|
|
await tx.query("COMMIT");
|
|
return result;
|
|
} catch (error) {
|
|
try { await tx.query("ROLLBACK"); } catch {}
|
|
throw error;
|
|
} finally {
|
|
tx.release?.();
|
|
}
|
|
}
|
|
await client.query("BEGIN");
|
|
try {
|
|
const result = await fn(client);
|
|
await client.query("COMMIT");
|
|
return result;
|
|
} catch (error) {
|
|
try { await client.query("ROLLBACK"); } catch {}
|
|
throw error;
|
|
}
|
|
} catch (error) {
|
|
this.recordRuntimeDbOperationFailure(label || "transaction", error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async getQueryClient() {
|
|
if (this.queryClient) {
|
|
return this.queryClient;
|
|
}
|
|
|
|
if (this.pool) {
|
|
return this.pool;
|
|
}
|
|
|
|
let pg;
|
|
try {
|
|
pg = await (this.pgModuleLoader ? this.pgModuleLoader() : import("pg"));
|
|
} catch (error) {
|
|
if (error?.code === "ERR_MODULE_NOT_FOUND") {
|
|
const driverError = new Error("Postgres runtime adapter requires the pg package");
|
|
driverError.code = "HWLAB_PG_DRIVER_MISSING";
|
|
throw driverError;
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
const Pool = pg.Pool ?? pg.default?.Pool;
|
|
if (typeof Pool !== "function") {
|
|
const driverError = new Error("Postgres runtime adapter could not load pg.Pool");
|
|
driverError.code = "HWLAB_PG_DRIVER_MISSING";
|
|
throw driverError;
|
|
}
|
|
|
|
this.pool = new Pool(buildPostgresPoolConfig({
|
|
dbUrl: this.dbUrl,
|
|
sslMode: this.sslMode,
|
|
timeoutMs: this.env?.HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS,
|
|
queryTimeoutMs: this.env?.[RUNTIME_DB_QUERY_TIMEOUT_MS_ENV],
|
|
poolMax: this.env?.[RUNTIME_DB_POOL_MAX_ENV]
|
|
}));
|
|
if (typeof this.pool.on === "function") {
|
|
this.pool.on("error", (error) => this.handlePostgresPoolError(error));
|
|
}
|
|
return this.pool;
|
|
}
|
|
|
|
handlePostgresPoolError(error) {
|
|
const classified = classifyRuntimeDbError(error);
|
|
this.lastReadiness = this.blockedSummary(classified);
|
|
this.logger?.warn?.({
|
|
event: "postgres_runtime_pool_error",
|
|
blocker: classified.blocker,
|
|
queryResult: classified.connection?.queryResult ?? "query_blocked",
|
|
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
|
|
retryable: classified.retryable === true,
|
|
transient: classified.transient === true,
|
|
retryAfterMs: classified.retryAfterMs ?? null,
|
|
valuesRedacted: true,
|
|
endpointRedacted: true
|
|
});
|
|
}
|
|
|
|
blockedSummary({ blocker, reason, schema, migration, connection, gates, retryable, transient, retryAfterMs }) {
|
|
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,
|
|
retryable: retryable === true ? true : undefined,
|
|
transient: transient === true ? true : undefined,
|
|
retryAfterMs: Number.isFinite(Number(retryAfterMs)) ? Math.max(0, Math.trunc(Number(retryAfterMs))) : undefined,
|
|
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()
|
|
});
|
|
}
|
|
}
|