diff --git a/package-lock.json b/package-lock.json index c642cf054..5eaa85a84 100644 --- a/package-lock.json +++ b/package-lock.json @@ -743,6 +743,10 @@ "resolved": "packages/harness-state", "link": true }, + "node_modules/@earendil-works/diffex-harness-trace": { + "resolved": "packages/harness-trace", + "link": true + }, "node_modules/@earendil-works/diffex-protocol": { "resolved": "packages/protocol", "link": true @@ -5732,6 +5736,39 @@ "node": ">=22.19.0" } }, + "packages/harness-trace": { + "name": "@earendil-works/diffex-harness-trace", + "version": "0.2.0", + "license": "MIT", + "dependencies": { + "@earendil-works/diffex-telemetry": "0.2.0" + }, + "devDependencies": { + "@types/node": "24.12.4", + "shx": "0.4.0", + "vitest": "4.1.9" + }, + "engines": { + "node": ">=22.19.0" + } + }, + "packages/harness-trace/node_modules/@types/node": { + "version": "24.12.4", + "resolved": "https://registry.npmjs.org/@types/node/-/node-24.12.4.tgz", + "integrity": "sha512-GUUEShf+PBCGW2KaXwcIt3Yk+e3pkKwWKb9GSyM9WQVE+ep2jzmHdGsHzu4wgcZy5fN9FBdVzjpBQsYlpfpgLA==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~7.16.0" + } + }, + "packages/harness-trace/node_modules/undici-types": { + "version": "7.16.0", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.16.0.tgz", + "integrity": "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==", + "dev": true, + "license": "MIT" + }, "packages/protocol": { "name": "@earendil-works/diffex-protocol", "version": "0.2.0", diff --git a/package.json b/package.json index e4cbe1721..9506df2b5 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,8 @@ ], "scripts": { "clean": "npm run clean --workspaces", - "build": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", - "build:offline": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", + "build": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", + "build:offline": "cd packages/harness-state && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../harness-trace && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../session-backends/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../client && npm run build && cd ../server && npm run build && cd ../coding-agent && npm run build", "check": "biome check --write --error-on-warnings . && npm run check:pinned-deps && npm run check:ts-imports && npm run check:shrinkwrap && npm run check:install-lock:coding-agent && tsgo --noEmit && npm run check:browser-smoke", "check:browser-smoke": "node scripts/check-browser-smoke.mjs", "check:pinned-deps": "node scripts/check-pinned-deps.mjs", diff --git a/packages/harness-trace/package.json b/packages/harness-trace/package.json new file mode 100644 index 000000000..a683f3a02 --- /dev/null +++ b/packages/harness-trace/package.json @@ -0,0 +1,42 @@ +{ + "name": "@earendil-works/diffex-harness-trace", + "version": "0.2.0", + "description": "Private durable harness trace recording for Diffex", + "private": true, + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "files": [ + "dist" + ], + "scripts": { + "clean": "shx rm -rf dist", + "build": "tsgo -p tsconfig.build.json", + "test": "vitest --run", + "prepublishOnly": "npm run clean && npm run build" + }, + "author": "Earendil Works", + "license": "MIT", + "repository": { + "type": "git", + "url": "git+https://github.com/diffexai/diffex.git", + "directory": "packages/harness-trace" + }, + "engines": { + "node": ">=22.19.0" + }, + "dependencies": { + "@earendil-works/diffex-telemetry": "0.2.0" + }, + "devDependencies": { + "@types/node": "24.12.4", + "shx": "0.4.0", + "vitest": "4.1.9" + } +} diff --git a/packages/harness-trace/src/codec.ts b/packages/harness-trace/src/codec.ts new file mode 100644 index 000000000..9079a4962 --- /dev/null +++ b/packages/harness-trace/src/codec.ts @@ -0,0 +1,412 @@ +import type { AttributeValue, SpanAttributes, SpanStatus } from "@earendil-works/diffex-telemetry"; +import { + type AgentHarnessSessionSegmentCursor, + type CodingAgentV3EntryCursor, + type HarnessActivationReason, + type SessionAttachmentReason, + type SessionEntryDigest, + type SessionSegmentCursor, + TRACE_SCHEMA_VERSION, + type TraceFooterRecord, + type TraceHarnessActivateRecord, + type TraceHeaderRecord, + type TraceHealth, + type TraceMetadata, + type TraceProducer, + type TraceRecord, + type TraceRuntimeIdentity, + type TraceSessionAttachRecord, + type TraceSessionSegmentRecord, + type TraceSpanEndRecord, + type TraceSpanEventRecord, + type TraceSpanStartRecord, + type TraceTelemetrySchemaVersions, +} from "./types.ts"; + +function invalidRecord(message: string): never { + throw new Error(`Invalid trace record: ${message}`); +} + +function recordValue(value: unknown, name: string): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) invalidRecord(`${name} must be an object`); + return value as Record; +} + +function exactKeys(value: Record, keys: readonly string[], name: string): void { + const expected = new Set(keys); + for (const key of Object.keys(value)) { + if (!expected.has(key)) invalidRecord(`${name} contains unknown field ${key}`); + } + for (const key of keys) { + if (!(key in value)) invalidRecord(`${name} is missing field ${key}`); + } +} + +function optionalKeys( + value: Record, + required: readonly string[], + optional: readonly string[], + name: string, +): void { + const expected = new Set([...required, ...optional]); + for (const key of Object.keys(value)) { + if (!expected.has(key)) invalidRecord(`${name} contains unknown field ${key}`); + } + for (const key of required) { + if (!(key in value)) invalidRecord(`${name} is missing field ${key}`); + } +} + +function stringValue(value: unknown, name: string): string { + if (typeof value !== "string") invalidRecord(`${name} must be a string`); + return value; +} + +function nonEmptyStringValue(value: unknown, name: string): string { + const text = stringValue(value, name); + if (text.length === 0) invalidRecord(`${name} must be non-empty`); + return text; +} + +function positiveIntegerValue(value: unknown, name: string): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 1) { + invalidRecord(`${name} must be a positive safe integer`); + } + return value; +} + +function nonNegativeIntegerValue(value: unknown, name: string): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) { + invalidRecord(`${name} must be a non-negative safe integer`); + } + return value; +} + +function timestampValue(value: unknown): string { + if (typeof value !== "string") invalidRecord("timestamp must be a string"); + const milliseconds = Date.parse(value); + if (!Number.isFinite(milliseconds) || new Date(milliseconds).toISOString() !== value) { + invalidRecord("timestamp must be a canonical ISO 8601 instant"); + } + return value; +} + +function attributeValue(value: unknown, name: string): AttributeValue { + if (typeof value === "string" || typeof value === "boolean") return value; + if (typeof value === "number" && Number.isFinite(value)) return value; + if (!Array.isArray(value)) invalidRecord(`${name} is not a telemetry primitive`); + if (value.every((item) => typeof item === "string")) return value; + if (value.every((item) => typeof item === "boolean")) return value; + if (value.every((item) => typeof item === "number" && Number.isFinite(item))) return value; + invalidRecord(`${name} is not a homogeneous telemetry primitive array`); +} + +function attributesValue(value: unknown, name: string): SpanAttributes { + const attributes = recordValue(value, name); + const parsed: SpanAttributes = {}; + for (const [key, attribute] of Object.entries(attributes)) parsed[key] = attributeValue(attribute, `${name}.${key}`); + return parsed; +} + +function statusValue(value: unknown): SpanStatus { + const status = recordValue(value, "status"); + if (status.status === "ok") { + exactKeys(status, ["status"], "status"); + return { status: "ok" }; + } + if (status.status !== "error") invalidRecord("status.status must be ok or error"); + optionalKeys(status, ["status"], ["error"], "status"); + if (status.error === undefined) return { status: "error" }; + const error = recordValue(status.error, "status.error"); + exactKeys(error, ["name", "message"], "status.error"); + return { + status: "error", + error: { + name: stringValue(error.name, "status.error.name"), + message: stringValue(error.message, "status.error.message"), + }, + }; +} + +function healthValue(value: unknown): TraceHealth { + const health = recordValue(value, "health"); + exactKeys(health, ["droppedRecords", "writerErrors", "writerDisabled", "closed"], "health"); + if (typeof health.writerDisabled !== "boolean" || typeof health.closed !== "boolean") { + invalidRecord("health flags must be boolean"); + } + return { + droppedRecords: nonNegativeIntegerValue(health.droppedRecords, "health.droppedRecords"), + writerErrors: nonNegativeIntegerValue(health.writerErrors, "health.writerErrors"), + writerDisabled: health.writerDisabled, + closed: health.closed, + }; +} + +function metadataValue(value: unknown): TraceMetadata { + const metadata = recordValue(value, "producer.metadata"); + const parsed: Record = {}; + for (const [key, item] of Object.entries(metadata)) { + if (item === null || typeof item === "string" || typeof item === "boolean") parsed[key] = item; + else if (typeof item === "number" && Number.isFinite(item)) parsed[key] = item; + else invalidRecord(`producer.metadata.${key} must be a JSON primitive`); + } + return parsed; +} + +function telemetrySchemaVersionsValue(value: unknown): TraceTelemetrySchemaVersions { + const versions = recordValue(value, "producer.telemetrySchemaVersions"); + const parsed: Record = {}; + for (const [name, version] of Object.entries(versions)) { + if (name.length === 0) invalidRecord("telemetry schema names must be non-empty"); + parsed[name] = positiveIntegerValue(version, `producer.telemetrySchemaVersions.${name}`); + } + return parsed; +} + +function runtimeValue(value: unknown): TraceRuntimeIdentity { + const runtime = recordValue(value, "producer.runtime"); + exactKeys(runtime, ["kind", "version"], "producer.runtime"); + return { + kind: nonEmptyStringValue(runtime.kind, "producer.runtime.kind"), + version: nonEmptyStringValue(runtime.version, "producer.runtime.version"), + }; +} + +function producerValue(value: unknown): TraceProducer { + const producer = recordValue(value, "producer"); + exactKeys( + producer, + ["diffexVersion", "sourceRevision", "runtime", "telemetrySchemaVersions", "metadata"], + "producer", + ); + return { + diffexVersion: nonEmptyStringValue(producer.diffexVersion, "producer.diffexVersion"), + sourceRevision: nonEmptyStringValue(producer.sourceRevision, "producer.sourceRevision"), + runtime: runtimeValue(producer.runtime), + telemetrySchemaVersions: telemetrySchemaVersionsValue(producer.telemetrySchemaVersions), + metadata: metadataValue(producer.metadata), + }; +} + +function commonRecord(value: Record, keys: readonly string[], name: string): void { + exactKeys(value, ["type", "sequence", "timestamp", ...keys], name); + positiveIntegerValue(value.sequence, "sequence"); + timestampValue(value.timestamp); +} + +function parseHeader(value: Record): TraceHeaderRecord { + commonRecord(value, ["schemaVersion", "traceId", "producer"], "trace_header"); + if (value.sequence !== 1) invalidRecord("trace_header sequence must be 1"); + if (value.schemaVersion !== TRACE_SCHEMA_VERSION) invalidRecord("unsupported schema version"); + return { + type: "trace_header", + sequence: 1, + timestamp: timestampValue(value.timestamp), + schemaVersion: TRACE_SCHEMA_VERSION, + traceId: nonEmptyStringValue(value.traceId, "traceId"), + producer: producerValue(value.producer), + }; +} + +const sessionAttachmentReasons = new Set([ + "startup", + "resume", + "new", + "import", + "fork", + "switch", +]); + +function sessionAttachmentReasonValue(value: unknown): SessionAttachmentReason { + const reason = nonEmptyStringValue(value, "reason") as SessionAttachmentReason; + if (!sessionAttachmentReasons.has(reason)) invalidRecord("session attachment reason is invalid"); + return reason; +} + +function parseSessionAttach(value: Record): TraceSessionAttachRecord { + commonRecord( + value, + ["bindingId", "sessionId", "path", "sessionFormat", "formatVersion", "reason"], + "session_attach", + ); + return { + type: "session_attach", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + bindingId: nonEmptyStringValue(value.bindingId, "bindingId"), + sessionId: nonEmptyStringValue(value.sessionId, "sessionId"), + path: nonEmptyStringValue(value.path, "path"), + sessionFormat: nonEmptyStringValue(value.sessionFormat, "sessionFormat"), + formatVersion: positiveIntegerValue(value.formatVersion, "formatVersion"), + reason: sessionAttachmentReasonValue(value.reason), + }; +} + +const harnessActivationReasons = new Set(["initial", "switch"]); + +function harnessActivationReasonValue(value: unknown): HarnessActivationReason { + const reason = nonEmptyStringValue(value, "reason") as HarnessActivationReason; + if (!harnessActivationReasons.has(reason)) invalidRecord("harness activation reason is invalid"); + return reason; +} + +function parseHarnessActivate(value: Record): TraceHarnessActivateRecord { + commonRecord(value, ["harnessVersion", "reason"], "harness_activate"); + return { + type: "harness_activate", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + harnessVersion: nonEmptyStringValue(value.harnessVersion, "harnessVersion"), + reason: harnessActivationReasonValue(value.reason), + }; +} + +function parseSpanStart(value: Record): TraceSpanStartRecord { + commonRecord(value, ["spanId", "parentSpanId", "name", "attributes"], "span_start"); + const parentSpanId = value.parentSpanId === null ? null : positiveIntegerValue(value.parentSpanId, "parentSpanId"); + return { + type: "span_start", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + spanId: positiveIntegerValue(value.spanId, "spanId"), + parentSpanId, + name: stringValue(value.name, "name"), + attributes: attributesValue(value.attributes, "attributes"), + }; +} + +function parseSpanEvent(value: Record): TraceSpanEventRecord { + commonRecord(value, ["spanId", "name", "attributes"], "span_event"); + return { + type: "span_event", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + spanId: positiveIntegerValue(value.spanId, "spanId"), + name: stringValue(value.name, "name"), + attributes: attributesValue(value.attributes, "attributes"), + }; +} + +function parseSpanEnd(value: Record): TraceSpanEndRecord { + commonRecord(value, ["spanId", "attributes", "status"], "span_end"); + return { + type: "span_end", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + spanId: positiveIntegerValue(value.spanId, "spanId"), + attributes: attributesValue(value.attributes, "attributes"), + status: statusValue(value.status), + }; +} + +function sessionEntryDigestValue(value: unknown, name: string): SessionEntryDigest { + const digest = recordValue(value, name); + exactKeys(digest, ["kind", "algorithm", "value"], name); + if (digest.kind !== "prefix" && digest.kind !== "content") invalidRecord(`${name}.kind is invalid`); + if (digest.algorithm !== "sha256") invalidRecord(`${name}.algorithm is invalid`); + const digestValue = nonEmptyStringValue(digest.value, `${name}.value`); + if (!/^[a-f0-9]{64}$/.test(digestValue)) invalidRecord(`${name}.value must be a lowercase SHA-256 digest`); + return { kind: digest.kind, algorithm: "sha256", value: digestValue }; +} + +function codingAgentEntryCursorValue(value: unknown, name: string): CodingAgentV3EntryCursor { + const cursor = recordValue(value, name); + exactKeys(cursor, ["entryOrdinal", "entryId", "digest"], name); + return { + entryOrdinal: nonNegativeIntegerValue(cursor.entryOrdinal, `${name}.entryOrdinal`), + entryId: nonEmptyStringValue(cursor.entryId, `${name}.entryId`), + digest: sessionEntryDigestValue(cursor.digest, `${name}.digest`), + }; +} + +function agentHarnessCursorValue(value: Record): AgentHarnessSessionSegmentCursor { + exactKeys(value, ["kind", "startSequence", "endSequence"], "cursor"); + const startSequence = positiveIntegerValue(value.startSequence, "cursor.startSequence"); + const endSequence = positiveIntegerValue(value.endSequence, "cursor.endSequence"); + if (endSequence < startSequence) invalidRecord("session segment ends before it starts"); + return { kind: "agent-harness", startSequence, endSequence }; +} + +function sessionSegmentCursorValue(value: unknown): SessionSegmentCursor { + const cursor = recordValue(value, "cursor"); + if (cursor.kind === "agent-harness") return agentHarnessCursorValue(cursor); + if (cursor.kind !== "coding-agent-v3") invalidRecord("cursor.kind is invalid"); + exactKeys(cursor, ["kind", "start", "end"], "cursor"); + const start = codingAgentEntryCursorValue(cursor.start, "cursor.start"); + const end = codingAgentEntryCursorValue(cursor.end, "cursor.end"); + if (end.entryOrdinal < start.entryOrdinal) invalidRecord("session segment ends before it starts"); + return { kind: "coding-agent-v3", start, end }; +} + +function parseSessionSegment(value: Record): TraceSessionSegmentRecord { + commonRecord(value, ["bindingId", "lane", "operationId", "cursor"], "session_segment"); + return { + type: "session_segment", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + bindingId: nonEmptyStringValue(value.bindingId, "bindingId"), + lane: nonEmptyStringValue(value.lane, "lane"), + operationId: nonEmptyStringValue(value.operationId, "operationId"), + cursor: sessionSegmentCursorValue(value.cursor), + }; +} + +function parseFooter(value: Record): TraceFooterRecord { + commonRecord(value, ["terminationStatus", "health"], "trace_footer"); + if ( + value.terminationStatus !== "completed" && + value.terminationStatus !== "failed" && + value.terminationStatus !== "aborted" + ) { + invalidRecord("terminationStatus is invalid"); + } + return { + type: "trace_footer", + sequence: positiveIntegerValue(value.sequence, "sequence"), + timestamp: timestampValue(value.timestamp), + terminationStatus: value.terminationStatus, + health: healthValue(value.health), + }; +} + +/** Encodes exactly one validated trace record with LF framing. */ +export function encodeTraceRecord(record: TraceRecord): string { + const encoded = JSON.stringify(record); + parseTraceRecord(encoded); + return `${encoded}\n`; +} + +/** Parses and validates exactly one trace record, with or without one trailing LF. */ +export function parseTraceRecord(line: string): TraceRecord { + const text = line.endsWith("\n") ? line.slice(0, -1) : line; + if (text.length === 0 || text.includes("\n") || text.includes("\r")) { + invalidRecord("expected one LF-framed JSON value"); + } + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch (error) { + throw new Error("Invalid trace record: malformed JSON", { cause: error }); + } + const value = recordValue(parsed, "record"); + switch (value.type) { + case "trace_header": + return parseHeader(value); + case "session_attach": + return parseSessionAttach(value); + case "harness_activate": + return parseHarnessActivate(value); + case "span_start": + return parseSpanStart(value); + case "span_event": + return parseSpanEvent(value); + case "span_end": + return parseSpanEnd(value); + case "session_segment": + return parseSessionSegment(value); + case "trace_footer": + return parseFooter(value); + default: + return invalidRecord("unknown record type"); + } +} diff --git a/packages/harness-trace/src/index.ts b/packages/harness-trace/src/index.ts new file mode 100644 index 000000000..bf68e5fd5 --- /dev/null +++ b/packages/harness-trace/src/index.ts @@ -0,0 +1,38 @@ +export { encodeTraceRecord, parseTraceRecord } from "./codec.ts"; +export { readJsonlTraceFile } from "./reader.ts"; +export { createJsonlTraceContext } from "./recorder.ts"; +export { + type AgentHarnessSessionSegmentCursor, + type CodingAgentV3EntryCursor, + type CodingAgentV3SessionSegmentCursor, + type CreateJsonlTraceContextOptions, + type HarnessActivation, + type HarnessActivationReason, + type JsonlTraceContext, + type SessionAttachment, + type SessionAttachmentReason, + type SessionEntryDigest, + type SessionEntryDigestKind, + type SessionSegment, + type SessionSegmentCursor, + TRACE_SCHEMA_VERSION, + type TraceDiagnostic, + type TraceDiagnosticCode, + type TraceFooterRecord, + type TraceHarnessActivateRecord, + type TraceHeaderRecord, + type TraceHealth, + type TraceMetadata, + type TraceMetadataValue, + type TraceProducer, + type TraceReadResult, + type TraceRecord, + type TraceRuntimeIdentity, + type TraceSessionAttachRecord, + type TraceSessionSegmentRecord, + type TraceSpanEndRecord, + type TraceSpanEventRecord, + type TraceSpanStartRecord, + type TraceTelemetrySchemaVersions, + type TraceTerminationStatus, +} from "./types.ts"; diff --git a/packages/harness-trace/src/reader.ts b/packages/harness-trace/src/reader.ts new file mode 100644 index 000000000..dca825a0e --- /dev/null +++ b/packages/harness-trace/src/reader.ts @@ -0,0 +1,230 @@ +import { createReadStream } from "node:fs"; +import { StringDecoder } from "node:string_decoder"; +import { parseTraceRecord } from "./codec.ts"; +import { + TRACE_SCHEMA_VERSION, + type TraceDiagnostic, + type TraceDiagnosticCode, + type TraceReadResult, + type TraceRecord, + type TraceSessionAttachRecord, +} from "./types.ts"; + +interface MutableSpanValidationState { + settled: boolean; +} + +interface RawRecord { + readonly type?: unknown; + readonly schemaVersion?: unknown; +} + +function rawRecord(line: string): RawRecord | undefined { + try { + const value: unknown = JSON.parse(line); + if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined; + return value as RawRecord; + } catch { + return undefined; + } +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : "Invalid trace record"; +} + +/** Streams, decodes, and validates a complete trace file without loading the file into memory. */ +export async function readJsonlTraceFile(path: string): Promise { + const records: TraceRecord[] = []; + const diagnostics: TraceDiagnostic[] = []; + const spans = new Map(); + const sessionBindings = new Map(); + let lineNumber = 0; + let expectedSequence = 1; + let headerSeen = false; + let footerSeen = false; + let schemaRejected = false; + let pending = ""; + const decoder = new StringDecoder("utf8"); + + const diagnose = (code: TraceDiagnosticCode, message: string, sequence?: number): void => { + diagnostics.push({ code, message, lineNumber, ...(sequence === undefined ? {} : { sequence }) }); + }; + + const validateRecord = (record: TraceRecord): void => { + if (record.sequence > expectedSequence) { + diagnose( + "sequence_gap", + `Expected trace sequence ${expectedSequence}, received ${record.sequence}`, + record.sequence, + ); + expectedSequence = record.sequence + 1; + } else if (record.sequence < expectedSequence) { + diagnose( + "sequence_out_of_order", + `Expected trace sequence ${expectedSequence}, received ${record.sequence}`, + record.sequence, + ); + } else { + expectedSequence++; + } + + if (footerSeen) { + diagnose("record_after_footer", `Record ${record.type} appears after trace closure`, record.sequence); + } + + switch (record.type) { + case "trace_header": + if (headerSeen) diagnose("duplicate_header", "Trace contains more than one header", record.sequence); + if (lineNumber !== 1) diagnose("misplaced_header", "Trace header is not the first record", record.sequence); + headerSeen = true; + break; + case "session_attach": + if (sessionBindings.has(record.bindingId)) { + diagnose( + "duplicate_session_binding", + `Session binding ${record.bindingId} is attached more than once`, + record.sequence, + ); + } else { + sessionBindings.set(record.bindingId, record); + } + break; + case "session_segment": { + const binding = sessionBindings.get(record.bindingId); + if (!binding) { + diagnose( + "unknown_session_binding", + `Session segment references unknown binding ${record.bindingId}`, + record.sequence, + ); + break; + } + const cursorMatches = + (record.cursor.kind === "coding-agent-v3" && + binding.sessionFormat === "coding-agent" && + binding.formatVersion === 3) || + (record.cursor.kind === "agent-harness" && binding.sessionFormat === "agent-harness"); + if (!cursorMatches) { + diagnose( + "session_cursor_mismatch", + `Session cursor ${record.cursor.kind} does not match ${binding.sessionFormat} v${binding.formatVersion}`, + record.sequence, + ); + } + break; + } + case "span_start": { + const parent = record.parentSpanId === null ? undefined : spans.get(record.parentSpanId); + if (spans.has(record.spanId)) { + diagnose("invalid_span_parentage", `Span ${record.spanId} starts more than once`, record.sequence); + } else { + spans.set(record.spanId, { settled: false }); + } + if (record.parentSpanId !== null && (!parent || parent.settled)) { + diagnose( + "invalid_span_parentage", + `Span ${record.spanId} references unavailable parent ${record.parentSpanId}`, + record.sequence, + ); + } + break; + } + case "span_event": { + const span = spans.get(record.spanId); + if (!span || span.settled) { + diagnose( + "invalid_span_parentage", + `Span event references unavailable span ${record.spanId}`, + record.sequence, + ); + } + break; + } + case "span_end": { + const span = spans.get(record.spanId); + if (!span) { + diagnose( + "invalid_span_parentage", + `Span settlement references unknown span ${record.spanId}`, + record.sequence, + ); + } else if (span.settled) { + diagnose("duplicate_span_settlement", `Span ${record.spanId} settles more than once`, record.sequence); + } else { + span.settled = true; + } + break; + } + case "trace_footer": + if (!headerSeen) diagnose("misplaced_footer", "Trace footer appears before its header", record.sequence); + if (footerSeen) diagnose("duplicate_footer", "Trace contains more than one footer", record.sequence); + footerSeen = true; + break; + case "harness_activate": + break; + } + }; + + const processLine = (line: string): void => { + lineNumber++; + if (schemaRejected) return; + const raw = rawRecord(line); + if ( + lineNumber === 1 && + raw?.type === "trace_header" && + typeof raw.schemaVersion === "number" && + Number.isSafeInteger(raw.schemaVersion) && + raw.schemaVersion > 0 && + raw.schemaVersion !== TRACE_SCHEMA_VERSION + ) { + diagnose( + "incompatible_schema_version", + `Trace schema version ${String(raw.schemaVersion)} is incompatible with supported version ${TRACE_SCHEMA_VERSION}`, + ); + schemaRejected = true; + return; + } + let record: TraceRecord; + try { + record = parseTraceRecord(line); + } catch (error) { + if (raw?.type === "trace_header") { + diagnose(headerSeen ? "duplicate_header" : "misplaced_header", errorMessage(error)); + } else { + diagnose("invalid_record", errorMessage(error)); + } + return; + } + records.push(record); + validateRecord(record); + }; + + for await (const chunk of createReadStream(path)) { + pending += decoder.write(chunk); + let newline = pending.indexOf("\n"); + while (newline !== -1) { + processLine(pending.slice(0, newline)); + pending = pending.slice(newline + 1); + newline = pending.indexOf("\n"); + } + } + pending += decoder.end(); + if (pending.length > 0) { + lineNumber++; + diagnostics.push({ + code: "partial_final_line", + message: "Trace ends with a partial final line", + lineNumber, + }); + } + + if (!schemaRejected) { + if (!headerSeen) + diagnostics.push({ code: "missing_header", message: "Trace header is missing", lineNumber: null }); + if (!footerSeen) + diagnostics.push({ code: "missing_footer", message: "Trace footer is missing", lineNumber: null }); + } + + return { records, diagnostics, closed: footerSeen }; +} diff --git a/packages/harness-trace/src/recorder.ts b/packages/harness-trace/src/recorder.ts new file mode 100644 index 000000000..e6a1746bc --- /dev/null +++ b/packages/harness-trace/src/recorder.ts @@ -0,0 +1,408 @@ +import { type FileHandle, mkdir, open } from "node:fs/promises"; +import { dirname } from "node:path"; +import { + type AttributeValue, + NOOP_TELEMETRY_CONTEXT, + type SpanAttributes, + type SpanOptions, + type SpanStatus, + type TelemetrySpan, +} from "@earendil-works/diffex-telemetry"; +import { encodeTraceRecord } from "./codec.ts"; +import { + type AgentHarnessSessionSegmentCursor, + type CodingAgentV3SessionSegmentCursor, + type CreateJsonlTraceContextOptions, + type HarnessActivation, + type JsonlTraceContext, + type SessionAttachment, + type SessionSegment, + type SessionSegmentCursor, + TRACE_SCHEMA_VERSION, + type TraceHealth, + type TraceMetadata, + type TraceRecord, + type TraceTelemetrySchemaVersions, + type TraceTerminationStatus, +} from "./types.ts"; + +interface MutableSpanState { + readonly id: number; + readonly parentId: number | null; + attributes: SpanAttributes; + status: SpanStatus; + explicitStatus: boolean; + settled: boolean; +} + +function copyAttributeValue(value: AttributeValue): AttributeValue { + if (typeof value === "number" && !Number.isFinite(value)) throw new Error("Telemetry numbers must be finite"); + if (!Array.isArray(value)) return value; + const copy = [...value] as AttributeValue; + if ((copy as readonly number[]).some((item) => typeof item === "number" && !Number.isFinite(item))) { + throw new Error("Telemetry numbers must be finite"); + } + return copy; +} + +function copyAttributes(attributes?: SpanAttributes): SpanAttributes { + const copy: SpanAttributes = {}; + if (!attributes) return copy; + for (const [name, value] of Object.entries(attributes)) { + if (value !== undefined) copy[name] = copyAttributeValue(value); + } + return copy; +} + +function mergeAttributes(current: SpanAttributes, attributes: SpanAttributes): SpanAttributes { + const merged = copyAttributes(current); + for (const [name, value] of Object.entries(attributes)) { + if (value !== undefined) merged[name] = copyAttributeValue(value); + } + return merged; +} + +function copyStatus(status: SpanStatus): SpanStatus { + if (status.status === "ok") return { status: "ok" }; + return status.error + ? { status: "error", error: { name: status.error.name, message: status.error.message } } + : { status: "error" }; +} + +function automaticErrorStatus(error: unknown): SpanStatus { + try { + if (error instanceof Error) { + return { status: "error", error: { name: error.name, message: error.message } }; + } + } catch { + // Error inspection is passive. Preserve only the failure classification. + } + return { status: "error" }; +} + +function copyMetadata( + metadata?: Readonly>, +): TraceMetadata { + const copy: Record = {}; + if (!metadata) return copy; + for (const [name, value] of Object.entries(metadata)) { + if (value === undefined) continue; + if (typeof value === "number" && !Number.isFinite(value)) + throw new Error("Trace metadata numbers must be finite"); + copy[name] = value; + } + return copy; +} + +function copyTelemetrySchemaVersions(versions: Readonly>): TraceTelemetrySchemaVersions { + return { ...versions }; +} + +function copySessionSegmentCursor(cursor: SessionSegmentCursor): SessionSegmentCursor { + if (cursor.kind === "agent-harness") { + return { + kind: "agent-harness", + startSequence: cursor.startSequence, + endSequence: cursor.endSequence, + } satisfies AgentHarnessSessionSegmentCursor; + } + return { + kind: "coding-agent-v3", + start: { + entryOrdinal: cursor.start.entryOrdinal, + entryId: cursor.start.entryId, + digest: { ...cursor.start.digest }, + }, + end: { + entryOrdinal: cursor.end.entryOrdinal, + entryId: cursor.end.entryId, + digest: { ...cursor.end.digest }, + }, + } satisfies CodingAgentV3SessionSegmentCursor; +} + +class JsonlTraceRecorder implements JsonlTraceContext { + readonly destinationPath: string; + readonly #file: FileHandle; + #nextSequence = 2; + #nextSpanId = 1; + #pendingWrite: Promise = Promise.resolve(); + #droppedRecords = 0; + #writerErrors = 0; + #writerDisabled = false; + #closed = false; + #closePromise: Promise | undefined; + + constructor(destinationPath: string, file: FileHandle) { + this.destinationPath = destinationPath; + this.#file = file; + } + + #timestamp(): string { + return new Date().toISOString(); + } + + #sequence(): number { + return this.#nextSequence++; + } + + #enqueue(record: TraceRecord): boolean { + if (this.#writerDisabled) { + this.#droppedRecords++; + return false; + } + let encoded: string; + try { + encoded = encodeTraceRecord(record); + } catch { + this.#droppedRecords++; + this.#writerErrors++; + return false; + } + this.#pendingWrite = this.#pendingWrite.then(async () => { + if (this.#writerDisabled) { + this.#droppedRecords++; + return; + } + try { + await this.#file.writeFile(encoded, { encoding: "utf8" }); + } catch { + this.#droppedRecords++; + this.#writerErrors++; + this.#writerDisabled = true; + } + }); + return true; + } + + #startSpan( + parent: MutableSpanState | undefined, + options: SpanOptions, + callback: (span: TelemetrySpan) => T | Promise, + ): Promise { + if (this.#closed || parent?.settled) return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + + let state: MutableSpanState; + let name: string; + try { + name = options.name; + state = { + id: this.#nextSpanId++, + parentId: parent?.id ?? null, + attributes: copyAttributes(options.attributes), + status: { status: "ok" }, + explicitStatus: false, + settled: false, + }; + const admitted = this.#enqueue({ + type: "span_start", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + spanId: state.id, + parentSpanId: state.parentId, + name, + attributes: copyAttributes(state.attributes), + }); + if (!admitted) return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + } catch { + return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + } + + const span: TelemetrySpan = { + startSpan: ( + childOptions: SpanOptions, + childCallback: (child: TelemetrySpan) => Result | Promise, + ) => this.#startSpan(state, childOptions, childCallback), + addEvent: (eventName, attributes) => { + if (state.settled || this.#closed) return; + try { + this.#enqueue({ + type: "span_event", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + spanId: state.id, + name: eventName, + attributes: copyAttributes(attributes), + }); + } catch { + // Recording calls remain passive for malformed or unreadable payloads. + } + }, + setAttributes: (attributes) => { + if (state.settled || this.#closed) return; + try { + state.attributes = mergeAttributes(state.attributes, attributes); + } catch { + // Attribute updates are atomic and passive. + } + }, + setStatus: (status) => { + if (state.settled || this.#closed) return; + try { + state.status = copyStatus(status); + state.explicitStatus = true; + } catch { + // Status updates are atomic and passive. + } + }, + }; + + const settle = (failed: boolean, error?: unknown): void => { + if (state.settled) return; + state.settled = true; + if (this.#closed) return; + if (failed && !state.explicitStatus) state.status = automaticErrorStatus(error); + this.#enqueue({ + type: "span_end", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + spanId: state.id, + attributes: copyAttributes(state.attributes), + status: copyStatus(state.status), + }); + }; + + let result: T | Promise; + try { + result = callback(span); + } catch (error) { + settle(true, error); + return Promise.reject(error); + } + + return Promise.resolve(result).then( + (value) => { + settle(false); + return value; + }, + (error: unknown) => { + settle(true, error); + throw error; + }, + ); + } + + startSpan(options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + return this.#startSpan(undefined, options, callback); + } + + attachSession(attachment: SessionAttachment): void { + if (this.#closed) return; + try { + this.#enqueue({ + type: "session_attach", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + bindingId: attachment.bindingId, + sessionId: attachment.sessionId, + path: attachment.path, + sessionFormat: attachment.sessionFormat, + formatVersion: attachment.formatVersion, + reason: attachment.reason, + }); + } catch { + // Session attachment recording is passive. + } + } + + activateHarness(activation: HarnessActivation): void { + if (this.#closed) return; + try { + this.#enqueue({ + type: "harness_activate", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + harnessVersion: activation.harnessVersion, + reason: activation.reason, + }); + } catch { + // Harness activation recording is passive. + } + } + + recordSessionSegment(segment: SessionSegment): void { + if (this.#closed) return; + try { + this.#enqueue({ + type: "session_segment", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + bindingId: segment.bindingId, + lane: segment.lane, + operationId: segment.operationId, + cursor: copySessionSegmentCursor(segment.cursor), + }); + } catch { + // Session segment recording is passive. + } + } + + getHealth(): TraceHealth { + return { + droppedRecords: this.#droppedRecords, + writerErrors: this.#writerErrors, + writerDisabled: this.#writerDisabled, + closed: this.#closed, + }; + } + + async flush(): Promise { + await this.#pendingWrite; + } + + close(terminationStatus: TraceTerminationStatus = "completed"): Promise { + if (this.#closePromise) return this.#closePromise; + this.#closed = true; + this.#enqueue({ + type: "trace_footer", + sequence: this.#sequence(), + timestamp: this.#timestamp(), + terminationStatus, + health: this.getHealth(), + }); + this.#closePromise = this.#finishClose(); + return this.#closePromise; + } + + async #finishClose(): Promise { + await this.#pendingWrite; + try { + await this.#file.close(); + } catch { + this.#writerErrors++; + this.#writerDisabled = true; + } + } + + [Symbol.asyncDispose](): Promise { + return this.close(); + } +} + +/** Creates a fresh trace file and writes its header before returning the passive recorder. */ +export async function createJsonlTraceContext(options: CreateJsonlTraceContextOptions): Promise { + const header: TraceRecord = { + type: "trace_header", + sequence: 1, + timestamp: new Date().toISOString(), + schemaVersion: TRACE_SCHEMA_VERSION, + traceId: options.traceId, + producer: { + diffexVersion: options.diffexVersion, + sourceRevision: options.sourceRevision, + runtime: { kind: options.runtimeKind, version: options.runtimeVersion }, + telemetrySchemaVersions: copyTelemetrySchemaVersions(options.telemetrySchemaVersions), + metadata: copyMetadata(options.metadata), + }, + }; + const encodedHeader = encodeTraceRecord(header); + await mkdir(dirname(options.destinationPath), { recursive: true, mode: 0o700 }); + const file = await open(options.destinationPath, "wx", 0o600); + try { + await file.writeFile(encodedHeader, { encoding: "utf8" }); + } catch (error) { + await file.close().catch(() => {}); + throw error; + } + return new JsonlTraceRecorder(options.destinationPath, file); +} diff --git a/packages/harness-trace/src/types.ts b/packages/harness-trace/src/types.ts new file mode 100644 index 000000000..27bd99251 --- /dev/null +++ b/packages/harness-trace/src/types.ts @@ -0,0 +1,203 @@ +import type { SpanAttributes, SpanStatus, TelemetryContext } from "@earendil-works/diffex-telemetry"; + +export const TRACE_SCHEMA_VERSION = 1 as const; + +export type TraceMetadataValue = string | number | boolean | null; + +export interface TraceMetadata { + readonly [name: string]: TraceMetadataValue; +} + +export interface TraceTelemetrySchemaVersions { + readonly [name: string]: number; +} + +export interface TraceRuntimeIdentity { + readonly kind: string; + readonly version: string; +} + +export interface TraceProducer { + readonly diffexVersion: string; + readonly sourceRevision: string; + readonly runtime: TraceRuntimeIdentity; + readonly telemetrySchemaVersions: TraceTelemetrySchemaVersions; + readonly metadata: TraceMetadata; +} + +export interface TraceHealth { + readonly droppedRecords: number; + readonly writerErrors: number; + readonly writerDisabled: boolean; + readonly closed: boolean; +} + +export type TraceTerminationStatus = "completed" | "failed" | "aborted"; + +export interface CreateJsonlTraceContextOptions { + readonly destinationPath: string; + readonly traceId: string; + readonly diffexVersion: string; + readonly sourceRevision: string; + readonly runtimeKind: string; + readonly runtimeVersion: string; + readonly telemetrySchemaVersions: Readonly>; + readonly metadata?: Readonly>; +} + +export type SessionAttachmentReason = "startup" | "resume" | "new" | "import" | "fork" | "switch"; + +export interface SessionAttachment { + readonly bindingId: string; + readonly sessionId: string; + readonly path: string; + readonly sessionFormat: string; + readonly formatVersion: number; + readonly reason: SessionAttachmentReason; +} + +export type HarnessActivationReason = "initial" | "switch"; + +export interface HarnessActivation { + readonly harnessVersion: string; + readonly reason: HarnessActivationReason; +} + +export type SessionEntryDigestKind = "prefix" | "content"; + +export interface SessionEntryDigest { + readonly kind: SessionEntryDigestKind; + readonly algorithm: "sha256"; + readonly value: string; +} + +export interface CodingAgentV3EntryCursor { + readonly entryOrdinal: number; + readonly entryId: string; + readonly digest: SessionEntryDigest; +} + +export interface CodingAgentV3SessionSegmentCursor { + readonly kind: "coding-agent-v3"; + readonly start: CodingAgentV3EntryCursor; + readonly end: CodingAgentV3EntryCursor; +} + +export interface AgentHarnessSessionSegmentCursor { + readonly kind: "agent-harness"; + readonly startSequence: number; + readonly endSequence: number; +} + +export type SessionSegmentCursor = CodingAgentV3SessionSegmentCursor | AgentHarnessSessionSegmentCursor; + +export interface SessionSegment { + readonly bindingId: string; + readonly lane: string; + readonly operationId: string; + readonly cursor: SessionSegmentCursor; +} + +export interface JsonlTraceContext extends TelemetryContext, AsyncDisposable { + readonly destinationPath: string; + attachSession(attachment: SessionAttachment): void; + activateHarness(activation: HarnessActivation): void; + recordSessionSegment(segment: SessionSegment): void; + getHealth(): TraceHealth; + flush(): Promise; + close(terminationStatus?: TraceTerminationStatus): Promise; +} + +interface SequencedTraceRecord { + readonly sequence: number; + readonly timestamp: string; +} + +export interface TraceHeaderRecord extends SequencedTraceRecord { + readonly type: "trace_header"; + readonly schemaVersion: typeof TRACE_SCHEMA_VERSION; + readonly traceId: string; + readonly producer: TraceProducer; +} + +export interface TraceSessionAttachRecord extends SequencedTraceRecord, SessionAttachment { + readonly type: "session_attach"; +} + +export interface TraceHarnessActivateRecord extends SequencedTraceRecord, HarnessActivation { + readonly type: "harness_activate"; +} + +export interface TraceSpanStartRecord extends SequencedTraceRecord { + readonly type: "span_start"; + readonly spanId: number; + readonly parentSpanId: number | null; + readonly name: string; + readonly attributes: Readonly; +} + +export interface TraceSpanEventRecord extends SequencedTraceRecord { + readonly type: "span_event"; + readonly spanId: number; + readonly name: string; + readonly attributes: Readonly; +} + +export interface TraceSpanEndRecord extends SequencedTraceRecord { + readonly type: "span_end"; + readonly spanId: number; + readonly attributes: Readonly; + readonly status: SpanStatus; +} + +export interface TraceSessionSegmentRecord extends SequencedTraceRecord, SessionSegment { + readonly type: "session_segment"; +} + +export interface TraceFooterRecord extends SequencedTraceRecord { + readonly type: "trace_footer"; + readonly terminationStatus: TraceTerminationStatus; + readonly health: TraceHealth; +} + +export type TraceDiagnosticCode = + | "invalid_record" + | "incompatible_schema_version" + | "sequence_gap" + | "sequence_out_of_order" + | "duplicate_header" + | "misplaced_header" + | "missing_header" + | "duplicate_footer" + | "misplaced_footer" + | "missing_footer" + | "record_after_footer" + | "invalid_span_parentage" + | "duplicate_span_settlement" + | "duplicate_session_binding" + | "unknown_session_binding" + | "session_cursor_mismatch" + | "partial_final_line"; + +export interface TraceDiagnostic { + readonly code: TraceDiagnosticCode; + readonly message: string; + readonly lineNumber: number | null; + readonly sequence?: number; +} + +export interface TraceReadResult { + readonly records: readonly TraceRecord[]; + readonly diagnostics: readonly TraceDiagnostic[]; + readonly closed: boolean; +} + +export type TraceRecord = + | TraceHeaderRecord + | TraceSessionAttachRecord + | TraceHarnessActivateRecord + | TraceSpanStartRecord + | TraceSpanEventRecord + | TraceSpanEndRecord + | TraceSessionSegmentRecord + | TraceFooterRecord; diff --git a/packages/harness-trace/test/fixtures/agent-harness.jsonl b/packages/harness-trace/test/fixtures/agent-harness.jsonl new file mode 100644 index 000000000..a2f476e56 --- /dev/null +++ b/packages/harness-trace/test/fixtures/agent-harness.jsonl @@ -0,0 +1,5 @@ +{"type":"trace_header","sequence":1,"timestamp":"2026-08-18T11:00:00.000Z","schemaVersion":1,"traceId":"agent-harness-trace","producer":{"diffexVersion":"0.2.0","sourceRevision":"abcdef","runtime":{"kind":"bun","version":"1.3.0"},"telemetrySchemaVersions":{"ai":1,"harness":1},"metadata":{}}} +{"type":"session_attach","sequence":2,"timestamp":"2026-08-18T11:00:00.001Z","bindingId":"harness-binding","sessionId":"harness-session","path":"/sessions/harness-session.db","sessionFormat":"agent-harness","formatVersion":1,"reason":"startup"} +{"type":"harness_activate","sequence":3,"timestamp":"2026-08-18T11:00:00.002Z","harnessVersion":"sha256:5678","reason":"initial"} +{"type":"session_segment","sequence":4,"timestamp":"2026-08-18T11:00:00.003Z","bindingId":"harness-binding","lane":"main","operationId":"operation-1","cursor":{"kind":"agent-harness","startSequence":40,"endSequence":49}} +{"type":"trace_footer","sequence":5,"timestamp":"2026-08-18T11:00:00.004Z","terminationStatus":"completed","health":{"droppedRecords":0,"writerErrors":0,"writerDisabled":false,"closed":true}} diff --git a/packages/harness-trace/test/fixtures/coding-agent-v3.jsonl b/packages/harness-trace/test/fixtures/coding-agent-v3.jsonl new file mode 100644 index 000000000..b8ce08bc9 --- /dev/null +++ b/packages/harness-trace/test/fixtures/coding-agent-v3.jsonl @@ -0,0 +1,5 @@ +{"type":"trace_header","sequence":1,"timestamp":"2026-08-18T10:00:00.000Z","schemaVersion":1,"traceId":"coding-agent-trace","producer":{"diffexVersion":"0.2.0","sourceRevision":"abcdef","runtime":{"kind":"node","version":"22.19.0"},"telemetrySchemaVersions":{"ai":1,"coding-agent":1},"metadata":{}}} +{"type":"session_attach","sequence":2,"timestamp":"2026-08-18T10:00:00.001Z","bindingId":"coding-binding","sessionId":"coding-session","path":"/sessions/coding-session.jsonl","sessionFormat":"coding-agent","formatVersion":3,"reason":"resume"} +{"type":"harness_activate","sequence":3,"timestamp":"2026-08-18T10:00:00.002Z","harnessVersion":"sha256:1234","reason":"initial"} +{"type":"session_segment","sequence":4,"timestamp":"2026-08-18T10:00:00.003Z","bindingId":"coding-binding","lane":"main","operationId":"operation-1","cursor":{"kind":"coding-agent-v3","start":{"entryOrdinal":4,"entryId":"entry-4","digest":{"kind":"prefix","algorithm":"sha256","value":"1111111111111111111111111111111111111111111111111111111111111111"}},"end":{"entryOrdinal":7,"entryId":"entry-7","digest":{"kind":"content","algorithm":"sha256","value":"2222222222222222222222222222222222222222222222222222222222222222"}}}} +{"type":"trace_footer","sequence":5,"timestamp":"2026-08-18T10:00:00.004Z","terminationStatus":"completed","health":{"droppedRecords":0,"writerErrors":0,"writerDisabled":false,"closed":true}} diff --git a/packages/harness-trace/test/harness-trace.test.ts b/packages/harness-trace/test/harness-trace.test.ts new file mode 100644 index 000000000..57ab721de --- /dev/null +++ b/packages/harness-trace/test/harness-trace.test.ts @@ -0,0 +1,621 @@ +import { type FileHandle, mkdir, mkdtemp, open, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { RecordedTelemetrySpan, SpanAttributes } from "@earendil-works/diffex-telemetry"; +import { + createTelemetryAdapterConformance, + type TelemetryAdapterFixture, +} from "@earendil-works/diffex-telemetry/testing"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + createJsonlTraceContext, + encodeTraceRecord, + type JsonlTraceContext, + parseTraceRecord, + readJsonlTraceFile, + type TraceFooterRecord, + type TraceHeaderRecord, + type TraceRecord, + type TraceSpanEndRecord, + type TraceSpanEventRecord, + type TraceSpanStartRecord, +} from "../src/index.ts"; + +const temporaryDirectories: string[] = []; + +async function temporaryTracePath(name = "trace.jsonl"): Promise { + const directory = await mkdtemp(join(tmpdir(), "diffex-harness-trace-")); + temporaryDirectories.push(directory); + return join(directory, "nested", name); +} + +async function createTrace(path: string): Promise { + return createJsonlTraceContext({ + destinationPath: path, + traceId: "trace-1", + diffexVersion: "0.2.0", + sourceRevision: "abcdef", + runtimeKind: "node", + runtimeVersion: "22.19.0", + telemetrySchemaVersions: { ai: 1, harness: 1 }, + metadata: { invocation: "test", attempt: 2, enabled: true, empty: null }, + }); +} + +async function readRecords(context: JsonlTraceContext, path: string): Promise { + await context.flush(); + const contents = await readFile(path, "utf8"); + return contents + .split("\n") + .filter((line) => line.length > 0) + .map(parseTraceRecord); +} + +function fixturePath(name: string): string { + return fileURLToPath(new URL(`./fixtures/${name}`, import.meta.url)); +} + +function traceHeader(): TraceHeaderRecord { + return { + type: "trace_header", + sequence: 1, + timestamp: "2026-08-18T10:00:00.000Z", + schemaVersion: 1, + traceId: "trace-reader", + producer: { + diffexVersion: "0.2.0", + sourceRevision: "abcdef", + runtime: { kind: "node", version: "22.19.0" }, + telemetrySchemaVersions: { ai: 1 }, + metadata: {}, + }, + }; +} + +function traceFooter(sequence: number): TraceFooterRecord { + return { + type: "trace_footer", + sequence, + timestamp: "2026-08-18T10:00:01.000Z", + terminationStatus: "completed", + health: { droppedRecords: 0, writerErrors: 0, writerDisabled: false, closed: true }, + }; +} + +async function writeTraceLines(path: string, lines: readonly string[]): Promise { + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, lines.join(""), "utf8"); +} + +function normalizedSpans(records: readonly TraceRecord[]): readonly RecordedTelemetrySpan[] { + const starts = records.filter((record): record is TraceSpanStartRecord => record.type === "span_start"); + const events = records.filter((record): record is TraceSpanEventRecord => record.type === "span_event"); + const ends = records.filter((record): record is TraceSpanEndRecord => record.type === "span_end"); + return starts.map((start) => { + const end = ends.find((candidate) => candidate.spanId === start.spanId); + return { + id: start.spanId, + parentId: start.parentSpanId, + name: start.name, + attributes: end?.attributes ?? start.attributes, + events: events + .filter((event) => event.spanId === start.spanId) + .map((event) => ({ name: event.name, attributes: event.attributes })), + status: end?.status ?? { status: "ok" }, + settled: end !== undefined, + ...(end === undefined ? {} : { endSequence: end.sequence }), + }; + }); +} + +afterEach(async () => { + vi.useRealTimers(); + await Promise.all( + temporaryDirectories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })), + ); +}); + +const conformance = createTelemetryAdapterConformance(async () => { + const path = await temporaryTracePath(); + const context = await createTrace(path); + return { + context, + async getSpans() { + return normalizedSpans(await readRecords(context, path)); + }, + [Symbol.asyncDispose]: () => context.close(), + } satisfies TelemetryAdapterFixture; +}); + +describe("JSONL telemetry adapter conformance", () => { + for (const group of new Set(conformance.map((testCase) => testCase.group))) { + describe(group, () => { + for (const testCase of conformance.filter((candidate) => candidate.group === group)) { + it(testCase.name, () => testCase.run()); + } + }); + } +}); + +describe("JSONL trace records", () => { + it("records ordered telemetry across session and harness transitions", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-08-18T10:00:00.000Z")); + const path = await temporaryTracePath(); + const context = await createTrace(path); + const automaticFailure = new Error("automatic failure"); + context.attachSession({ + bindingId: "binding-1", + sessionId: "session-1", + path: "/sessions/session-1.jsonl", + sessionFormat: "coding-agent", + formatVersion: 3, + reason: "startup", + }); + context.activateHarness({ harnessVersion: "sha256:1234", reason: "initial" }); + + const parentResult = context.startSpan( + { name: "parent", attributes: { retained: "start", overwritten: "start", omitted: undefined } }, + async (parent) => { + vi.setSystemTime(new Date("2026-08-18T10:00:01.000Z")); + parent.addEvent("checkpoint", { tags: ["one", "two"], omitted: undefined }); + parent.setAttributes({ overwritten: "end", count: 3 }); + const first = parent.startSpan({ name: "first" }, async () => { + await Promise.resolve(); + throw automaticFailure; + }); + const second = parent.startSpan({ name: "second" }, (span) => { + span.setStatus({ status: "error", error: { name: "Expected", message: "declined" } }); + return "second-result"; + }); + expect(await second).toBe("second-result"); + await expect(first).rejects.toBe(automaticFailure); + return "parent-result"; + }, + ); + expect(await parentResult).toBe("parent-result"); + vi.setSystemTime(new Date("2026-08-18T10:00:02.000Z")); + context.recordSessionSegment({ + bindingId: "binding-1", + lane: "main", + operationId: "operation-1", + cursor: { + kind: "coding-agent-v3", + start: { + entryOrdinal: 12, + entryId: "entry-12", + digest: { kind: "prefix", algorithm: "sha256", value: "1".repeat(64) }, + }, + end: { + entryOrdinal: 19, + entryId: "entry-19", + digest: { kind: "content", algorithm: "sha256", value: "2".repeat(64) }, + }, + }, + }); + context.attachSession({ + bindingId: "binding-2", + sessionId: "session-2", + path: "/sessions/session-2.db", + sessionFormat: "agent-harness", + formatVersion: 1, + reason: "switch", + }); + context.activateHarness({ harnessVersion: "sha256:5678", reason: "switch" }); + context.recordSessionSegment({ + bindingId: "binding-2", + lane: "main", + operationId: "operation-2", + cursor: { kind: "agent-harness", startSequence: 20, endSequence: 27 }, + }); + await context.close("completed"); + + const records = await readRecords(context, path); + expect(records.map((record) => record.sequence)).toEqual(records.map((_, index) => index + 1)); + expect(records.map((record) => record.type)).toEqual([ + "trace_header", + "session_attach", + "harness_activate", + "span_start", + "span_event", + "span_start", + "span_start", + "span_end", + "span_end", + "span_end", + "session_segment", + "session_attach", + "harness_activate", + "session_segment", + "trace_footer", + ]); + const header = records[0]; + expect(header).toMatchObject({ + type: "trace_header", + timestamp: "2026-08-18T10:00:00.000Z", + traceId: "trace-1", + producer: { + diffexVersion: "0.2.0", + sourceRevision: "abcdef", + runtime: { kind: "node", version: "22.19.0" }, + telemetrySchemaVersions: { ai: 1, harness: 1 }, + metadata: { invocation: "test", attempt: 2, enabled: true, empty: null }, + }, + }); + expect(records.filter((record) => record.type === "session_attach")).toEqual([ + expect.objectContaining({ bindingId: "binding-1", sessionId: "session-1", reason: "startup" }), + expect.objectContaining({ bindingId: "binding-2", sessionId: "session-2", reason: "switch" }), + ]); + expect(records.filter((record) => record.type === "harness_activate")).toEqual([ + expect.objectContaining({ harnessVersion: "sha256:1234", reason: "initial" }), + expect.objectContaining({ harnessVersion: "sha256:5678", reason: "switch" }), + ]); + const [parentStart, firstStart, secondStart] = records.filter( + (record): record is TraceSpanStartRecord => record.type === "span_start", + ); + expect(parentStart?.parentSpanId).toBeNull(); + expect(firstStart?.parentSpanId).toBe(parentStart?.spanId); + expect(secondStart?.parentSpanId).toBe(parentStart?.spanId); + const event = records.find((record): record is TraceSpanEventRecord => record.type === "span_event"); + expect(event).toMatchObject({ + timestamp: "2026-08-18T10:00:01.000Z", + spanId: parentStart?.spanId, + name: "checkpoint", + attributes: { tags: ["one", "two"] }, + }); + const ends = records.filter((record): record is TraceSpanEndRecord => record.type === "span_end"); + expect(ends.find((record) => record.spanId === parentStart?.spanId)).toMatchObject({ + attributes: { retained: "start", overwritten: "end", count: 3 }, + status: { status: "ok" }, + }); + expect(ends.find((record) => record.spanId === firstStart?.spanId)?.status).toEqual({ + status: "error", + error: { name: "Error", message: "automatic failure" }, + }); + expect(ends.find((record) => record.spanId === secondStart?.spanId)?.status).toEqual({ + status: "error", + error: { name: "Expected", message: "declined" }, + }); + expect(records.at(-2)).toMatchObject({ + type: "session_segment", + timestamp: "2026-08-18T10:00:02.000Z", + bindingId: "binding-2", + lane: "main", + operationId: "operation-2", + cursor: { kind: "agent-harness", startSequence: 20, endSequence: 27 }, + }); + expect(records.at(-1)).toMatchObject({ + type: "trace_footer", + terminationStatus: "completed", + health: { droppedRecords: 0, writerErrors: 0, writerDisabled: false, closed: true }, + }); + }); + + it("copies attributes and metadata without retaining mutable inputs", async () => { + const path = await temporaryTracePath(); + const metadata: Record = { source: "original", omitted: undefined }; + const context = await createJsonlTraceContext({ + destinationPath: path, + traceId: "trace-copy", + diffexVersion: "0.2.0", + sourceRevision: "revision", + runtimeKind: "node", + runtimeVersion: "22.19.0", + telemetrySchemaVersions: { ai: 1 }, + metadata, + }); + metadata.source = "mutated"; + const tags = ["original"]; + const attributes: SpanAttributes = { tags }; + await context.startSpan({ name: "copy", attributes }, (span) => { + tags[0] = "mutated"; + (attributes.tags as string[])[0] = "mutated again"; + span.addEvent("copy-event", { tags }); + tags[0] = "late mutation"; + }); + await context.close(); + + const records = await readRecords(context, path); + expect(records[0]).toMatchObject({ producer: { metadata: { source: "original" } } }); + expect(records.find((record) => record.type === "span_start")).toMatchObject({ + attributes: { tags: ["original"] }, + }); + expect(records.find((record) => record.type === "span_event")).toMatchObject({ + attributes: { tags: ["mutated again"] }, + }); + }); + + it("writes only one footer when close is repeated", async () => { + const path = await temporaryTracePath(); + const context = await createTrace(path); + const firstClose = context.close("aborted"); + expect(context.close("completed")).toBe(firstClose); + await firstClose; + await context.close("failed"); + + const records = await readRecords(context, path); + expect(records.filter((record) => record.type === "trace_footer")).toEqual([ + expect.objectContaining({ terminationStatus: "aborted" }), + ]); + }); + + it("opens destinations exclusively and creates parent directories", async () => { + const path = await temporaryTracePath(); + const first = await createTrace(path); + await expect(createTrace(path)).rejects.toMatchObject({ code: "EEXIST" }); + await first.close(); + }); + + it("leaves a missing footer as an incomplete invocation marker", async () => { + const path = await temporaryTracePath(); + const context = await createTrace(path); + await context.startSpan({ name: "unfinished-invocation" }, () => "done"); + + const incompleteRecords = await readRecords(context, path); + expect(incompleteRecords.at(-1)?.type).toBe("span_end"); + expect(incompleteRecords.some((record) => record.type === "trace_footer")).toBe(false); + await context.close(); + }); + + it("suppresses serialization failures without disabling later writes", async () => { + const path = await temporaryTracePath(); + const context = await createTrace(path); + const result = await context.startSpan({ name: "serialization" }, (span) => { + span.addEvent(1n as unknown as string); + span.addEvent("retained", { value: true }); + return 42; + }); + expect(result).toBe(42); + await context.flush(); + expect(context.getHealth()).toMatchObject({ droppedRecords: 1, writerErrors: 1, writerDisabled: false }); + await context.close(); + + const records = await readRecords(context, path); + expect(records.filter((record) => record.type === "span_event")).toEqual([ + expect.objectContaining({ name: "retained", attributes: { value: true } }), + ]); + }); + + it("suppresses fatal writes, disables the writer, and preserves callback behavior", async () => { + const path = await temporaryTracePath(); + const context = await createTrace(path); + const probePath = await temporaryTracePath("probe"); + await mkdir(dirname(probePath), { recursive: true }); + const probe = await open(probePath, "wx"); + const fileHandlePrototype = Object.getPrototypeOf(probe) as { writeFile: FileHandleWrite }; + await probe.close(); + const writeFailure = vi + .spyOn(fileHandlePrototype, "writeFile") + .mockRejectedValue(new Error("simulated write failure")); + + const expected = { preserved: true }; + const result = context.startSpan({ name: "write-failure" }, (span) => { + span.addEvent("queued-after-start"); + return expected; + }); + expect(await result).toBe(expected); + await context.flush(); + expect(context.getHealth()).toMatchObject({ droppedRecords: 3, writerErrors: 1, writerDisabled: true }); + writeFailure.mockRestore(); + await context.close("failed"); + + const records = await readRecords(context, path); + expect(records.map((record) => record.type)).toEqual(["trace_header"]); + const readResult = await readJsonlTraceFile(path); + expect(readResult.records).toEqual(records); + expect(readResult.diagnostics.map((diagnostic) => diagnostic.code)).toContain("missing_footer"); + }); +}); + +type FileHandleWrite = FileHandle["writeFile"]; + +describe("single-record codec", () => { + it("round trips one LF-framed record", () => { + const record: TraceRecord = { + type: "span_event", + sequence: 2, + timestamp: "2026-08-18T10:00:00.000Z", + spanId: 1, + name: "event", + attributes: { value: true }, + }; + expect(parseTraceRecord(encodeTraceRecord(record))).toEqual(record); + }); + + it.each([ + ["malformed JSON", "{"], + ["multiple records", "{}\n{}\n"], + ["unknown record", '{"type":"unknown"}'], + [ + "unsupported schema", + JSON.stringify({ + type: "trace_header", + sequence: 1, + timestamp: "2026-08-18T10:00:00.000Z", + schemaVersion: 2, + traceId: "trace", + producer: { + diffexVersion: "1", + sourceRevision: "r", + runtime: { kind: "node", version: "22" }, + telemetrySchemaVersions: {}, + metadata: {}, + }, + }), + ], + [ + "invalid segment", + JSON.stringify({ + type: "session_segment", + sequence: 2, + timestamp: "2026-08-18T10:00:00.000Z", + bindingId: "binding", + lane: "main", + operationId: "operation", + cursor: { kind: "agent-harness", startSequence: 3, endSequence: 2 }, + }), + ], + [ + "unknown field", + JSON.stringify({ + type: "span_event", + sequence: 2, + timestamp: "2026-08-18T10:00:00.000Z", + spanId: 1, + name: "event", + attributes: {}, + payload: "not allowed", + }), + ], + ])("rejects %s", (_name, line) => { + expect(() => parseTraceRecord(line)).toThrow(/invalid trace record/i); + }); +}); + +describe("whole-file reader", () => { + it.each([ + ["coding-agent-v3.jsonl", "coding-agent-v3"], + ["agent-harness.jsonl", "agent-harness"], + ])("validates the %s cursor fixture", async (name, cursorKind) => { + const result = await readJsonlTraceFile(fixturePath(name)); + expect(result.diagnostics).toEqual([]); + expect(result.closed).toBe(true); + expect(result.records.find((record) => record.type === "session_segment")).toMatchObject({ + cursor: { kind: cursorKind }, + }); + }); + + it("detects sequence gaps while preserving parsed records", async () => { + const path = await temporaryTracePath(); + await writeTraceLines(path, [ + encodeTraceRecord(traceHeader()), + encodeTraceRecord({ + type: "harness_activate", + sequence: 3, + timestamp: "2026-08-18T10:00:00.500Z", + harnessVersion: "sha256:1234", + reason: "initial", + }), + encodeTraceRecord(traceFooter(4)), + ]); + + const result = await readJsonlTraceFile(path); + expect(result.records).toHaveLength(3); + expect(result.diagnostics).toEqual([expect.objectContaining({ code: "sequence_gap", sequence: 3 })]); + }); + + it.each([ + [ + "duplicate_header", + [encodeTraceRecord(traceHeader()), encodeTraceRecord(traceHeader()), encodeTraceRecord(traceFooter(3))], + ], + [ + "misplaced_header", + [ + encodeTraceRecord({ + type: "harness_activate", + sequence: 1, + timestamp: "2026-08-18T10:00:00.000Z", + harnessVersion: "sha256:1234", + reason: "initial", + }), + encodeTraceRecord(traceHeader()), + encodeTraceRecord(traceFooter(3)), + ], + ], + [ + "duplicate_footer", + [encodeTraceRecord(traceHeader()), encodeTraceRecord(traceFooter(2)), encodeTraceRecord(traceFooter(3))], + ], + ["misplaced_footer", [encodeTraceRecord(traceFooter(1)), encodeTraceRecord(traceHeader())]], + ] as const)("detects %s", async (diagnosticCode, lines) => { + const path = await temporaryTracePath(); + await writeTraceLines(path, lines); + const result = await readJsonlTraceFile(path); + expect(result.diagnostics.map((diagnostic) => diagnostic.code)).toContain(diagnosticCode); + }); + + it("detects invalid parentage and duplicate span settlement", async () => { + const path = await temporaryTracePath(); + await writeTraceLines(path, [ + encodeTraceRecord(traceHeader()), + encodeTraceRecord({ + type: "span_start", + sequence: 2, + timestamp: "2026-08-18T10:00:00.100Z", + spanId: 1, + parentSpanId: 99, + name: "orphan", + attributes: {}, + }), + encodeTraceRecord({ + type: "span_end", + sequence: 3, + timestamp: "2026-08-18T10:00:00.200Z", + spanId: 1, + attributes: {}, + status: { status: "ok" }, + }), + encodeTraceRecord({ + type: "span_end", + sequence: 4, + timestamp: "2026-08-18T10:00:00.300Z", + spanId: 1, + attributes: {}, + status: { status: "ok" }, + }), + encodeTraceRecord(traceFooter(5)), + ]); + + const result = await readJsonlTraceFile(path); + expect(result.diagnostics.map((diagnostic) => diagnostic.code)).toEqual([ + "invalid_span_parentage", + "duplicate_span_settlement", + ]); + }); + + it("detects records after closure", async () => { + const path = await temporaryTracePath(); + await writeTraceLines(path, [ + encodeTraceRecord(traceHeader()), + encodeTraceRecord(traceFooter(2)), + encodeTraceRecord({ + type: "harness_activate", + sequence: 3, + timestamp: "2026-08-18T10:00:02.000Z", + harnessVersion: "sha256:late", + reason: "switch", + }), + ]); + + const result = await readJsonlTraceFile(path); + expect(result.records).toHaveLength(3); + expect(result.diagnostics).toEqual([expect.objectContaining({ code: "record_after_footer", sequence: 3 })]); + }); + + it("detects a partial final line and missing footer", async () => { + const path = await temporaryTracePath(); + await writeTraceLines(path, [encodeTraceRecord(traceHeader()), '{"type":"trace_footer"']); + + const result = await readJsonlTraceFile(path); + expect(result.records).toEqual([traceHeader()]); + expect(result.diagnostics.map((diagnostic) => diagnostic.code)).toEqual(["partial_final_line", "missing_footer"]); + expect(result.closed).toBe(false); + }); + + it("rejects a forward schema without interpreting later records", async () => { + const path = await temporaryTracePath(); + await writeTraceLines(path, [ + `${JSON.stringify({ ...traceHeader(), schemaVersion: 2 })}\n`, + encodeTraceRecord(traceFooter(2)), + ]); + + const result = await readJsonlTraceFile(path); + expect(result.records).toEqual([]); + expect(result.diagnostics).toEqual([ + expect.objectContaining({ code: "incompatible_schema_version", lineNumber: 1 }), + ]); + expect(result.closed).toBe(false); + }); +}); diff --git a/packages/harness-trace/tsconfig.build.json b/packages/harness-trace/tsconfig.build.json new file mode 100644 index 000000000..695dd9ad2 --- /dev/null +++ b/packages/harness-trace/tsconfig.build.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src" + }, + "include": ["src/**/*.ts"], + "exclude": ["node_modules", "dist", "**/*.d.ts", "src/**/*.d.ts"] +} diff --git a/tsconfig.json b/tsconfig.json index 4b90fa29a..aaebe66d1 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -7,6 +7,7 @@ "paths": { "*": ["./*"], "@earendil-works/diffex-harness-state": ["./packages/harness-state/src/index.ts"], + "@earendil-works/diffex-harness-trace": ["./packages/harness-trace/src/index.ts"], "@earendil-works/diffex-telemetry": ["./packages/telemetry/src/index.ts"], "@earendil-works/diffex-telemetry/testing": ["./packages/telemetry/src/testing/index.ts"], "@earendil-works/diffex-ai": ["./packages/ai/src/index.ts"],