diff --git a/apps/server/src/attachmentStore.ts b/apps/server/src/attachmentStore.ts index 5d8ab2e90181b..616b0d90c17b2 100644 --- a/apps/server/src/attachmentStore.ts +++ b/apps/server/src/attachmentStore.ts @@ -13,7 +13,7 @@ import { import { inferImageExtension, SAFE_IMAGE_FILE_EXTENSIONS } from "./imageMime.ts"; const ATTACHMENT_FILENAME_EXTENSIONS = [...SAFE_IMAGE_FILE_EXTENSIONS, ".bin"]; -const ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS = 80; +export const ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS = 80; const ATTACHMENT_ID_THREAD_SEGMENT_PATTERN = "[a-z0-9_]+(?:-[a-z0-9_]+)*"; const ATTACHMENT_ID_UUID_PATTERN = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"; const ATTACHMENT_ID_FILE_EXTENSION_PATTERN = "[a-z0-9]{1,10}"; diff --git a/apps/server/src/provider/EventNdjsonLogger.test.ts b/apps/server/src/provider/EventNdjsonLogger.test.ts index c04150dce8df6..3cb7870f3f518 100644 --- a/apps/server/src/provider/EventNdjsonLogger.test.ts +++ b/apps/server/src/provider/EventNdjsonLogger.test.ts @@ -156,6 +156,39 @@ it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => { }), ); + it.effect("keeps long thread ids that share a prefix in separate files", () => + Effect.gen(function* () { + const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-")); + const basePath = NodePath.join(tempDir, "events.log"); + const prefix = + "thread:delegated-task:command%3Amcp%3A00000000-0000-4000-8000-000000000000%3Adelegate-task%3A"; + + try { + const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }); + const logger = store.logger("native"); + yield* logger.write({ id: "first" }, ThreadId.make(`${prefix}review-the-diff`)); + yield* logger.write({ id: "second" }, ThreadId.make(`${prefix}write-the-tests`)); + yield* store.close(); + + const files = NodeFS.readdirSync(tempDir); + assert.equal(files.length, 2); + assert.deepEqual( + files + .map((name) => + NodeFS.readFileSync(NodePath.join(tempDir, name), "utf8") + .trim() + .split("\n") + .map((line) => parseLogLine(line).payload), + ) + .toSorted(), + [['{"id":"first"}'], ['{"id":"second"}']], + ); + } finally { + NodeFS.rmSync(tempDir, { recursive: true, force: true }); + } + }), + ); + it.effect("keeps shared store views non-owning when one adapter closes", () => Effect.gen(function* () { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-")); diff --git a/apps/server/src/provider/EventNdjsonLogger.ts b/apps/server/src/provider/EventNdjsonLogger.ts index 0e8c9043dcf0f..6a1d3844b2373 100644 --- a/apps/server/src/provider/EventNdjsonLogger.ts +++ b/apps/server/src/provider/EventNdjsonLogger.ts @@ -1,5 +1,6 @@ // @effect-diagnostics nodeBuiltinImport:off /** Native and canonical views share write state to avoid races on the same thread file. */ +import * as NodeCrypto from "node:crypto"; import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; @@ -18,7 +19,10 @@ import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as SynchronizedRef from "effect/SynchronizedRef"; -import { toSafeThreadAttachmentSegment } from "../attachmentStore.ts"; +import { + ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS, + toSafeThreadAttachmentSegment, +} from "../attachmentStore.ts"; import type * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts"; const MEBIBYTE = 1024 * 1024; @@ -145,8 +149,13 @@ function logWarning(message: string, context: Record): Effect.E } function resolveThreadSegment(raw: string | null | undefined): string { - const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null; - return normalized ?? GLOBAL_THREAD_SEGMENT; + if (typeof raw !== "string") return GLOBAL_THREAD_SEGMENT; + const normalized = toSafeThreadAttachmentSegment(raw); + if (normalized === null) return GLOBAL_THREAD_SEGMENT; + // Delegated thread ids share a prefix longer than the segment, so ids that may be cut get a hash. + if (raw.length <= ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS) return normalized; + const hash = NodeCrypto.createHash("sha256").update(raw).digest("hex").slice(0, 16); + return `${normalized}-${hash}`; } function resolveStreamLabel(stream: EventNdjsonStream): string { @@ -725,12 +734,13 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); const line = `[${observedAt}] ${resolveStreamLabel(stream)}: ${payload}\n`; const bytes = Buffer.byteLength(line); + const threadSegment = resolveThreadSegment(threadId); const action = yield* SynchronizedRef.modifyEffect(stateRef, (state) => { if (state.closed) { return Effect.succeed([{ flush: false }, state] as const); } const pending = state.pending; - pending.push({ stream, threadSegment: resolveThreadSegment(threadId), line, bytes }); + pending.push({ stream, threadSegment, line, bytes }); const pendingBytes = state.pendingBytes + bytes; const flush = resolved.batchWindowMs === 0 ||