root
1776 lines
92 KiB
TypeScript
1776 lines
92 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,
|
|
CLOUD_TRANSACTIONAL_REALTIME_MIGRATION_ID,
|
|
CLOUD_TRANSACTIONAL_REALTIME_SCHEMA_VERSION,
|
|
CLOUD_TRANSACTIONAL_REALTIME_TABLE_COLUMNS
|
|
} from "./schema.ts";
|
|
|
|
|
|
import { CloudRuntimeStore } from "./runtime-store-memory.ts";
|
|
import * as runtimeCore from "./runtime-store-core.ts";
|
|
import {
|
|
claimHwlabKafkaOutbox as claimHwlabKafkaOutboxOperation,
|
|
commitAgentRunKafkaProjection as commitAgentRunKafkaProjectionOperation,
|
|
completeHwlabKafkaOutbox as completeHwlabKafkaOutboxOperation,
|
|
hwlabKafkaProjectorStatus as hwlabKafkaProjectorStatusOperation,
|
|
readAtomicWorkbenchProjectionSync as readAtomicWorkbenchProjectionSyncOperation,
|
|
recordFailedAgentRunKafkaMessage as recordFailedAgentRunKafkaMessageOperation,
|
|
recordIgnoredAgentRunKafkaMessage as recordIgnoredAgentRunKafkaMessageOperation,
|
|
retryHwlabKafkaOutbox as retryHwlabKafkaOutboxOperation
|
|
} from "./runtime-store-postgres-kafka.ts";
|
|
import { subscribeWorkbenchProjectionCommits as subscribeWorkbenchProjectionCommitsOperation } from "./runtime-store-postgres-notify.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: immutableProjectionOutboxPayload("messages", fact, { role: fact.role, status: fact.status, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq }),
|
|
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: immutableProjectionOutboxPayload("traceEvents", fact, { type: fact.eventType, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq }),
|
|
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: immutableProjectionOutboxPayload("turns", fact, { status: fact.status, failureKind: fact.failureKind, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null }),
|
|
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 writeWorkbenchSessionAdmissionFact(params = {}, requestMeta = {}) {
|
|
const readiness = this.summary();
|
|
if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
|
|
await this.assertReadyForWrites();
|
|
}
|
|
const facts = normalizeWorkbenchFacts({ sessions: [params.fact ?? params.session ?? params] }, requestMeta, this.now());
|
|
const fact = facts.sessions[0];
|
|
if (!fact?.sessionId) throw new Error("Workbench session admission fact requires sessionId.");
|
|
await this.withDurableTransaction("workbench.session-admission.write", (client) => this.persistWorkbenchSessionAdmissionFact(fact, client));
|
|
const memoryResult = this.memory.writeWorkbenchSessionAdmissionFact({ fact }, requestMeta);
|
|
return withPersistence({ written: true, admissionOnly: true, fact: memoryResult.fact, valuesPrinted: false }, this.summary());
|
|
}
|
|
|
|
async promoteWorkbenchTurnAdmission(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 sessionFact = facts.sessions[0];
|
|
const inputFact = facts.inputs[0];
|
|
const turnFact = facts.turns[0];
|
|
if (!sessionFact?.sessionId || !inputFact?.inputId || !turnFact?.turnId) throw new Error("Workbench turn admission promotion requires session, input, and turn facts.");
|
|
const aggregateEvents = normalizeWorkbenchAggregateEventsForFacts(facts, requestMeta, this.now());
|
|
const transaction = await this.withDurableTransaction("workbench.turn-admission.promote", async (client) => {
|
|
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${turnFact.traceId}`]);
|
|
const existingTurnResult = await client.query(
|
|
"SELECT projected_seq, sealed FROM workbench_turns WHERE turn_id = $1 FOR UPDATE",
|
|
[turnFact.turnId]
|
|
);
|
|
const existingTurn = existingTurnResult.rows?.[0] ?? null;
|
|
const supersededByProjection = nonNegativeInteger(existingTurn?.projected_seq) > 0 || existingTurn?.sealed === true;
|
|
const persistedEvents = [];
|
|
const transactionEvents = supersededByProjection
|
|
? aggregateEvents.filter((event) => event.factFamily === "inputs")
|
|
: aggregateEvents;
|
|
for (const event of transactionEvents) persistedEvents.push(await this.persistWorkbenchAggregateEvent(event, client));
|
|
const eventIndex = indexWorkbenchAggregateEvents(persistedEvents);
|
|
const promotedInput = inputFactWithAggregateSeq(inputFact, eventIndex);
|
|
await this.persistWorkbenchSessionInputFact(promotedInput, client);
|
|
if (supersededByProjection) {
|
|
return { persistedEvents, promotedInput, outbox: null, supersededByProjection: true };
|
|
}
|
|
await this.persistWorkbenchSessionAdmissionFact(sessionFact, client);
|
|
for (const fact of facts.messages) await this.persistWorkbenchMessageAdmissionFact(fact, client);
|
|
for (const fact of facts.parts) await this.persistWorkbenchPartAdmissionFact(fact, client);
|
|
for (const fact of facts.turns) await this.persistWorkbenchTurnAdmissionFact(fact, client);
|
|
const turnEvent = eventIndex.get(`turn:${turnFact.turnId}`) ?? eventIndex.get(`sourceEvent:${turnFact.sourceEventId}`) ?? null;
|
|
const outbox = {
|
|
entityFamily: "turns",
|
|
entityId: turnFact.turnId,
|
|
eventSeq: turnEvent?.eventSeq ?? null,
|
|
aggregateId: turnEvent?.aggregateId ?? null,
|
|
aggregateSeq: turnEvent?.aggregateSeq ?? 0,
|
|
projectionRevision: turnEvent?.projectionRevision ?? turnFact.projectedSeq,
|
|
traceId: turnFact.traceId,
|
|
sessionId: turnFact.sessionId,
|
|
turnId: turnFact.turnId,
|
|
messageId: turnFact.messageId,
|
|
projectedSeq: turnFact.projectedSeq,
|
|
sourceSeq: turnFact.sourceSeq,
|
|
sourceEventId: turnFact.sourceEventId,
|
|
commitType: "admission",
|
|
terminal: false,
|
|
sealed: false,
|
|
payload: immutableProjectionOutboxPayload("turns", turnFact, { status: turnFact.status, admissionState: "promoted", eventSeq: turnEvent?.eventSeq ?? null, aggregateSeq: turnEvent?.aggregateSeq ?? null }),
|
|
createdAt: turnFact.updatedAt ?? this.now()
|
|
};
|
|
await this.persistWorkbenchProjectionOutbox(outbox, client);
|
|
return { persistedEvents, promotedInput, outbox, supersededByProjection: false };
|
|
});
|
|
const memoryResult = transaction.supersededByProjection
|
|
? this.memory.writeWorkbenchFacts({ facts: { inputs: [transaction.promotedInput] } }, requestMeta)
|
|
: this.memory.promoteWorkbenchTurnAdmission({ facts: { ...facts, inputs: [transaction.promotedInput] } }, requestMeta);
|
|
return withPersistence({
|
|
written: true,
|
|
promoted: true,
|
|
outboxCommitted: !transaction.supersededByProjection,
|
|
outbox: transaction.outbox,
|
|
supersededByProjection: transaction.supersededByProjection,
|
|
facts: memoryResult.facts,
|
|
events: transaction.persistedEvents,
|
|
valuesPrinted: false
|
|
}, this.summary());
|
|
}
|
|
|
|
async commitAgentRunKafkaProjection(params = {}) {
|
|
await this.assertReadyForWrites();
|
|
return commitAgentRunKafkaProjectionOperation(this, params);
|
|
}
|
|
|
|
async recordFailedAgentRunKafkaMessage(params = {}) {
|
|
await this.assertReadyForWrites();
|
|
return recordFailedAgentRunKafkaMessageOperation(this, params);
|
|
}
|
|
|
|
async recordIgnoredAgentRunKafkaMessage(params = {}) {
|
|
await this.assertReadyForWrites();
|
|
return recordIgnoredAgentRunKafkaMessageOperation(this, params);
|
|
}
|
|
|
|
async claimHwlabKafkaOutbox(params = {}) {
|
|
await this.assertReadyForWrites();
|
|
return claimHwlabKafkaOutboxOperation(this, params);
|
|
}
|
|
|
|
async completeHwlabKafkaOutbox(item = {}) {
|
|
return completeHwlabKafkaOutboxOperation(this, item);
|
|
}
|
|
|
|
async retryHwlabKafkaOutbox(item = {}, error = null, retryAt = null) {
|
|
return retryHwlabKafkaOutboxOperation(this, item, error, retryAt);
|
|
}
|
|
|
|
async hwlabKafkaProjectorStatus() {
|
|
await this.assertReadyForDurableReads("hwlab.kafka-projector.status");
|
|
return hwlabKafkaProjectorStatusOperation(this);
|
|
}
|
|
|
|
async readAtomicWorkbenchProjectionSync(params = {}) {
|
|
await this.assertReadyForDurableReads("workbench.projection-sync.read");
|
|
return readAtomicWorkbenchProjectionSyncOperation(this, params);
|
|
}
|
|
|
|
async subscribeWorkbenchProjectionCommits(listener) {
|
|
await this.assertReadyForDurableReads("workbench.projection-notify.listen");
|
|
return subscribeWorkbenchProjectionCommitsOperation(this, listener);
|
|
}
|
|
|
|
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 = COALESCE(EXCLUDED.owner_user_id, workbench_sessions.owner_user_id), project_id = COALESCE(EXCLUDED.project_id, workbench_sessions.project_id), conversation_id = COALESCE(EXCLUDED.conversation_id, workbench_sessions.conversation_id), thread_id = COALESCE(EXCLUDED.thread_id, workbench_sessions.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 persistWorkbenchSessionAdmissionFact(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 = COALESCE(workbench_sessions.owner_user_id, EXCLUDED.owner_user_id), project_id = COALESCE(workbench_sessions.project_id, EXCLUDED.project_id), conversation_id = COALESCE(workbench_sessions.conversation_id, EXCLUDED.conversation_id), thread_id = COALESCE(workbench_sessions.thread_id, EXCLUDED.thread_id), status = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.status ELSE workbench_sessions.status END, last_trace_id = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN COALESCE(EXCLUDED.last_trace_id, workbench_sessions.last_trace_id) ELSE workbench_sessions.last_trace_id END, session_json = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.session_json ELSE workbench_sessions.session_json END, updated_at = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.updated_at ELSE workbench_sessions.updated_at END",
|
|
[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 persistWorkbenchMessageAdmissionFact(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 = CASE WHEN workbench_messages.projected_seq <= 0 AND NOT workbench_messages.sealed THEN EXCLUDED.status ELSE workbench_messages.status END, source_event_id = CASE WHEN workbench_messages.projected_seq <= 0 AND NOT workbench_messages.sealed THEN EXCLUDED.source_event_id ELSE workbench_messages.source_event_id END, message_json = CASE WHEN workbench_messages.projected_seq <= 0 AND NOT workbench_messages.sealed THEN EXCLUDED.message_json ELSE workbench_messages.message_json END, updated_at = CASE WHEN workbench_messages.projected_seq <= 0 AND NOT workbench_messages.sealed THEN EXCLUDED.updated_at ELSE workbench_messages.updated_at END",
|
|
[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 persistWorkbenchPartAdmissionFact(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 = CASE WHEN workbench_parts.projected_seq <= 0 AND NOT workbench_parts.sealed THEN EXCLUDED.status ELSE workbench_parts.status END, source_event_id = CASE WHEN workbench_parts.projected_seq <= 0 AND NOT workbench_parts.sealed THEN EXCLUDED.source_event_id ELSE workbench_parts.source_event_id END, part_json = CASE WHEN workbench_parts.projected_seq <= 0 AND NOT workbench_parts.sealed THEN EXCLUDED.part_json ELSE workbench_parts.part_json END, updated_at = CASE WHEN workbench_parts.projected_seq <= 0 AND NOT workbench_parts.sealed THEN EXCLUDED.updated_at ELSE workbench_parts.updated_at END",
|
|
[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 persistWorkbenchTurnAdmissionFact(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 = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.message_id ELSE workbench_turns.message_id END, status = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.status ELSE workbench_turns.status END, source_event_id = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.source_event_id ELSE workbench_turns.source_event_id END, diagnostic_json = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.diagnostic_json ELSE workbench_turns.diagnostic_json END, turn_json = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.turn_json ELSE workbench_turns.turn_json END, updated_at = CASE WHEN workbench_turns.projected_seq <= 0 AND NOT workbench_turns.sealed THEN EXCLUDED.updated_at ELSE workbench_turns.updated_at END",
|
|
[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 (outbox_event_id, entity_family, entity_id, 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,$17,$18,$19) ON CONFLICT (outbox_event_id) DO NOTHING",
|
|
[record.outboxEventId ?? projectionOutboxEventId(record), record.entityFamily ?? projectionOutboxEntity(record).family, record.entityId ?? projectionOutboxEntity(record).id, 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, outbox_event_id, entity_family, entity_id, 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),
|
|
outboxEventId: row.outbox_event_id,
|
|
entityFamily: row.entity_family,
|
|
entityId: row.entity_id,
|
|
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 workbenchTransactionalRealtimeReadiness() {
|
|
const schemaResult = 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(CLOUD_TRANSACTIONAL_REALTIME_TABLE_COLUMNS)]
|
|
);
|
|
const schema = summarizeRuntimeSchema(schemaResult?.rows ?? [], CLOUD_TRANSACTIONAL_REALTIME_TABLE_COLUMNS);
|
|
const migrationResult = await this.query(
|
|
`SELECT id, schema_version FROM ${CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE} WHERE id = $1 LIMIT 1`,
|
|
[CLOUD_TRANSACTIONAL_REALTIME_MIGRATION_ID]
|
|
);
|
|
const row = migrationResult.rows?.[0] ?? null;
|
|
const migration = {
|
|
checked: true,
|
|
ready: row?.id === CLOUD_TRANSACTIONAL_REALTIME_MIGRATION_ID && row?.schema_version === CLOUD_TRANSACTIONAL_REALTIME_SCHEMA_VERSION,
|
|
table: CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE,
|
|
requiredMigrationId: CLOUD_TRANSACTIONAL_REALTIME_MIGRATION_ID,
|
|
requiredSchemaVersion: CLOUD_TRANSACTIONAL_REALTIME_SCHEMA_VERSION,
|
|
appliedMigrationId: row?.id ?? null,
|
|
appliedSchemaVersion: row?.schema_version ?? null,
|
|
missing: !row
|
|
};
|
|
return {
|
|
ready: schema.ready && migration.ready,
|
|
schema,
|
|
migration,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
async assertWorkbenchTransactionalRealtimeReady() {
|
|
const readiness = await this.workbenchTransactionalRealtimeReadiness();
|
|
if (readiness.ready) return readiness;
|
|
const error = new Error("Workbench transactional realtime schema or migration is not ready.");
|
|
error.code = readiness.schema.ready ? "workbench_transactional_realtime_migration_blocked" : "workbench_transactional_realtime_schema_blocked";
|
|
error.readiness = readiness;
|
|
error.valuesRedacted = true;
|
|
throw error;
|
|
}
|
|
|
|
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()
|
|
});
|
|
}
|
|
}
|
|
|
|
function projectionOutboxEventId(record = {}) {
|
|
const sourceEventId = textOr(record.sourceEventId, "");
|
|
const commitType = textOr(record.commitType, "event");
|
|
const family = textOr(record.entityFamily ?? record.payload?.family, "") || (commitType === "message" ? "messages" : record.payload?.type ? "traceEvents" : "turns");
|
|
const entityId = family === "traceEvents"
|
|
? textOr(record.entityId ?? record.payload?.fact?.id ?? record.aggregateId ?? record.sourceEventId ?? record.traceId, "projection")
|
|
: textOr(record.entityId ?? record.payload?.fact?.messageId ?? record.payload?.fact?.turnId ?? record.messageId ?? record.turnId ?? record.traceId ?? record.aggregateId, "projection");
|
|
if (sourceEventId) return `${sourceEventId}:${family}:${entityId}:${commitType}`;
|
|
const derived = stableJson({ family, entityId, commitType, eventSeq: record.eventSeq ?? null, aggregateId: record.aggregateId ?? null, aggregateSeq: record.aggregateSeq ?? 0, projectionRevision: record.projectionRevision ?? record.projectedSeq ?? 0, traceId: record.traceId ?? null, sessionId: record.sessionId ?? null });
|
|
return `derived:${createHash("sha256").update(derived).digest("hex")}`;
|
|
}
|
|
|
|
function projectionOutboxEntity(record = {}) {
|
|
const commitType = textOr(record.commitType, "event");
|
|
const family = textOr(record.entityFamily ?? record.payload?.family, "") || (commitType === "message" ? "messages" : record.payload?.type ? "traceEvents" : "turns");
|
|
const id = family === "traceEvents"
|
|
? textOr(record.entityId ?? record.payload?.fact?.id ?? record.aggregateId ?? record.sourceEventId ?? record.traceId, "projection")
|
|
: textOr(record.entityId ?? record.payload?.fact?.messageId ?? record.payload?.fact?.turnId ?? record.messageId ?? record.turnId ?? record.traceId ?? record.aggregateId, "projection");
|
|
return { family, id };
|
|
}
|
|
|
|
function immutableProjectionOutboxPayload(family, fact, metadata = {}) {
|
|
return { family, fact, metadata: { ...metadata, valuesRedacted: true }, valuesRedacted: true };
|
|
}
|