Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/server/src/attachmentStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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}";
Expand Down
33 changes: 33 additions & 0 deletions apps/server/src/provider/EventNdjsonLogger.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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-"));
Expand Down
18 changes: 14 additions & 4 deletions apps/server/src/provider/EventNdjsonLogger.ts
Original file line number Diff line number Diff line change
@@ -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";

Expand All @@ -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;
Expand Down Expand Up @@ -145,8 +149,13 @@ function logWarning(message: string, context: Record<string, unknown>): 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 {
Expand Down Expand Up @@ -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 ||
Expand Down
Loading