diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0b4e0e050e..3407d7a218 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1,3 +1,4 @@ +import * as ProofFs from "node:fs"; import type { OrchestrationV2ThreadLinkMetadata, GuestShareId, @@ -14173,6 +14174,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // dispatchWithReceiptEffect owns the dispatch span; this wrapper selects the thread locks. const dispatchWithReceipt = Effect.fnUntraced(function* (command: OrchestrationV2ServerCommand) { + if (command.type === "message.dispatch" && command.threadId.startsWith("proof24335:")) { + yield* Effect.sync(() => { + if (command.messageId === "proof24335:user") (globalThis as any).proof24335GateRead = true; + ProofFs.appendFileSync(process.env.PROOF_TRACE!, JSON.stringify({kind:"intake",commandId:command.commandId,messageId:command.messageId,threadId:command.threadId,createdBy:command.createdBy,creationSource:command.creationSource,at:new Date().toISOString()})+"\n"); + }); + if (command.createdBy === "agent") yield* Effect.promise(() => fetch(process.env.PROOF_GATE + "/agent").then(() => undefined)); + } const dispatch = threadDispatch.withLock( commandThreadId(command), dispatchWithReceiptEffect(command), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 0c4976dc90..f7d54a9186 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3,7 +3,7 @@ import { pendingGuestMessageCount } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; -import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as NodeSqliteClient from "../../../../packages/shared/src/nodeSqliteClient.ts"; import { makeThreadFind } from "./ThreadFind.ts"; import type { OrchestrationV2ThreadLinkMetadata, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index de12768ac7..7223942c10 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -757,6 +757,9 @@ const make = Effect.gen(function* () { const dispatch: ThreadManagementServiceShape["dispatch"] = (command) => ensureCommandTranscripts(command).pipe( Effect.andThen(assertDispatchTargetEnabled(command)), + Effect.andThen(command.type === "message.dispatch" && command.createdBy === "agent" && command.threadId.startsWith("proof24335:") + ? Effect.promise(() => fetch(process.env.PROOF_GATE + "/producer-ready").then(() => undefined)) + : Effect.void), Effect.andThen(orchestrator.dispatch(command)), ); diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 38913376cd..954e4cbc5d 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -3,7 +3,7 @@ import * as Layer from "effect/Layer"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; -import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as NodeSqliteClient from "../../../../packages/shared/src/nodeSqliteClient.ts"; import { runMigrations } from "./Migrations.ts"; import { initializeV2Database } from "./initializeV2Database.ts"; diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 4c2c2fe967..943ef27e58 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -226,10 +226,15 @@ const prepare = (sql, cached) => { statements.set(sql, statement); return statement; }; -parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { +parentPort.on("message", async ({ id, sql, params, cached, values, raw, safeIntegers, proofGated, proofGate, proofTrace }) => { let operation = "open"; try { if (db === undefined) throw new Error("database reader is unavailable"); + if (proofGated) { + require("node:fs").appendFileSync(proofTrace, JSON.stringify({kind:"sqlite-reader-query-pending",id,sql,params,db:workerData.filename,at:new Date().toISOString()})+"\\n"); + await fetch(proofGate + "/lineage"); + require("node:fs").appendFileSync(proofTrace, JSON.stringify({kind:"sqlite-reader-query-release",id,at:new Date().toISOString()})+"\\n"); + } operation = "prepare"; const statement = prepare(sql, cached); operation = "execute"; @@ -391,7 +396,9 @@ const makeReader = Effect.fnUntraced(function* ( if (pending.size === 1) worker.ref(); const cached = input.cached && !hasLargeBinding(input.params); // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. - worker.postMessage({ id, safeIntegers, ...input, cached }); + const proofGated = Boolean((globalThis as any).proof24335GateRead) && input.sql.includes("SELECT payload_json FROM orchestration_v2_projection_threads") && input.params.includes("proof24335:thread"); + if (proofGated) (globalThis as any).proof24335GateRead = false; + worker.postMessage({ id, safeIntegers, ...input, cached, proofGated, proofGate: process.env.PROOF_GATE, proofTrace: process.env.PROOF_TRACE }); }); });