diff --git a/.agents/skills/verify-session-lifecycle/coverage.json b/.agents/skills/verify-session-lifecycle/coverage.json index 91f0e7ef..702b5a94 100644 --- a/.agents/skills/verify-session-lifecycle/coverage.json +++ b/.agents/skills/verify-session-lifecycle/coverage.json @@ -133,7 +133,7 @@ "src/host/sessionState.ts#SessionState.requestDestroy": "f66074e815527ff1", "src/host/sessionState.ts#SessionState.recordDestroyed": "495b3a20c56b6330", "src/host/sessionState.ts#SessionState.recordExit": "a4f3f6706ae953c3", - "src/host/eventLog.ts#EventLog.open": "4d4a1c04ce17c37d", + "src/host/eventLog.ts#EventLog.open": "19229956e761db8c", "src/host/rpcServer.ts#RpcServer.listen": "aa62e1cd89eceaa4", "src/host/rpcServer.ts#RpcServer.close": "2d8092cceb4b8e13", "src/storage/manifests.ts#readValidatedJsonFile": "6287eb627b09bf80", diff --git a/docs/TROUBLESHOOTING.md b/docs/TROUBLESHOOTING.md index 319825e9..a2421fbc 100644 --- a/docs/TROUBLESHOOTING.md +++ b/docs/TROUBLESHOOTING.md @@ -101,9 +101,11 @@ To deliberately keep the marker, pass `--env PROMPT_EOL_MARK='%B%S%#%s%b'` (zsh' Each session records its terminal output in `events.jsonl`. The log stops at the first of two limits: 52,428,800 bytes (50 MB), or 250,000 events. A program that prints a lot reaches the byte limit. A program that prints in many tiny chunks reaches the event limit long before 50 MB. When the log stops, the host writes one marker and after it records only the program's exit. The marker reads `agent-tty: event log limit reached (52428800 bytes); later events are not recorded` when the byte limit stopped the log, and `agent-tty: event log limit reached (250000 events); later events are not recorded` when the event limit did. The program keeps running. Its later output, input, resizes and signals are not recorded, so snapshots, screenshots and recordings show the session as it was at the marker. An asciicast export ends with the marker. `mark` refuses a label that starts with `agent-tty: event log limit reached` with `INVALID_INPUT`, because only the host writes that marker. +To check whether a session's log stopped, run `agent-tty inspect --json`. The result has `eventLogStop`, `{ "reason": "bytes" | "events" | "unknown", "seq": }`, only when the log stopped. `reason` is the limit that stopped it, and `seq` is the `seq` of the marker. `unknown` means that the marker is in the log but its label names no limit, as in a log that agent-tty did not write. Without `--json`, `inspect` prints a line such as `Event Log Stopped: at seq 41 (byte limit); later events are not recorded`, and `(reason unknown)` for `unknown`. `inspect` reads the field from `events.jsonl` in the same read that counts `eventCount`, so it works for a live session and for an exited one, and `eventLogStop.seq` is never above `lastEventSeq`. + Commands on a live session read the stored log too, so after the marker they see the session as it was at the marker: -- `wait --text` and `wait --regex` for text that appears later time out. +- `wait --text` and `wait --regex` for text that appears later time out. A render wait that times out on a stopped log says so, in its message and in `error.details.eventLogStop`: `The event log stopped at seq 41 (byte limit), so output after it is not seen.` A `batch` `wait` step adds the same sentence. On an exited session, `wait` replays the stored log, and a wait that the replayed screen does not satisfy fails with `REPLAY_ERROR`. Its message ends with the same sentence, and `error.details.eventLogStop` carries the same facts. A CLI older than the host that runs the session (for example after a downgrade) cannot read that timeout result, and reports `RPC_ERROR` instead of `WAIT_TIMEOUT`. - `wait --screen-stable-ms` can report a stable screen while the program keeps printing, because the stored screen no longer changes. - `mouse` encodes against the mouse modes in the stored log. A mode that the program turned on after the marker is not known, so the event succeeds with `reported: false` and writes nothing. A mode that the program turned off after the marker is still on in the stored log, so the program receives mouse bytes that it no longer expects. - `wait --idle-ms` and `wait --exit` watch the live program and are not affected. diff --git a/docs/USAGE.md b/docs/USAGE.md index 268b0745..582a98cb 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -150,6 +150,8 @@ Useful flags: On timeout, a standalone `wait` exits `11` and prints an error envelope: `ok: false`, `error.code: "WAIT_TIMEOUT"`, and the wait result under `result` (`timedOut: true`, and `matched: false` for render waits). Without `--json`, it prints `WAIT_TIMEOUT: ` to stderr. When the wait lasted its whole timeout, the message names it, for example `Wait timed out after 30000 ms (the default; pass --timeout , or --timeout 0 to wait without a limit).` Inside `batch`, a timed-out `wait` step is a step failure with the same `WAIT_TIMEOUT` exit code under fail-fast, and its message names the step's timeout, for example `Render wait at step 0 timed out after 1000 ms.` +A render wait reads the session's event log. When that log has stopped at its limit (see [the event log limit](TROUBLESHOOTING.md#a-recording-ends-early-or-an-export-names-the-event-log-limit)), a timed-out render wait adds `The event log stopped at seq 41 (byte limit), so output after it is not seen.` to its message, with and without `--json`, and `error.details.eventLogStop` holds `{ "reason": "bytes", "seq": 41 }`. The host reports the stop with its timeout result, so a wait that times out at the moment the log stops names it too. A `batch` `wait` step adds the same sentence to its message. On an exited session, `wait` replays the stored log, and a wait that the replayed screen does not satisfy fails with `REPLAY_ERROR`. Its message ends with the same sentence, and `error.details.eventLogStop` carries the same facts. `reason` is `unknown` when the stop marker names no limit, and the sentence then reads `(reason unknown)`. `--idle-ms` and `--exit` waits do not read the log, so they add nothing. A wait that matches is not changed. + `WAIT_TIMEOUT` only means that the timeout ran out. When 10 render polls in a row fail (about 3 s, for example because the renderer cannot start), a render `wait` stops with the renderer's error instead, also with `--timeout 0`: the same `error.code` and exit code that `snapshot` gives for that failure, usually `RPC_ERROR` (exit `9`, `retryable: true`), and no `result`. A `batch` `wait` step fails with that code, and its step record has no `timedOut`. Fewer failed polls in a row are retried. ### Echo-Match @@ -325,6 +327,16 @@ agent-tty record diff --at-seq-a 0 --json - The line diff is LCS-minimal for any realistic screen. As a bounded-memory safeguard, when the differing region (after matching the common prefix and suffix) is so large that the product of its two side lengths exceeds a 4,000,000-cell budget (for example 1,000 × 4,001 lines), that region degrades to a plain delete-then-add block instead of a minimal diff — `identical`, both `screenHash` values, and full-screen reconstruction remain exact; only diff minimality is reduced. - The command works entirely offline from `events.jsonl`; sessions may be running or exited. The exit code is `0` whether or not the screens differ — automation should read `identical` from the JSON result. +## `inspect` + +```bash +agent-tty inspect --json +``` + +`inspect` prints the session record and what the Home holds for it: `eventCount`, `lastEventSeq`, `eventLogBytes`, `artifacts`, `terminationCategory`, `rendererRuntime` and, for a live host, `host`. + +`eventLogStop` appears only when the session's event log stopped at its limit (see [the event log limit](TROUBLESHOOTING.md#a-recording-ends-early-or-an-export-names-the-event-log-limit)). It is `{ "reason": "bytes" | "events" | "unknown", "seq": }`: `reason` names the limit that stopped the log (52,428,800 bytes, or 250,000 events, or `unknown` when the marker names no limit), and `seq` is the `seq` of the marker that the host wrote when it stopped. Events after that `seq` are not recorded, so snapshots, waits and exports show the session as it was then. `inspect` reads `eventCount`, `eventLogBytes` and the field in one read of `events.jsonl`, so they describe the same bytes, and a live session and an exited one report the same. It reads logs that are over the size limit that other commands read. The field is absent for a log that has not stopped. Without `--json`, one line reports it, for example `Event Log Stopped: at seq 41 (byte limit); later events are not recorded`. + ## `destroy` `destroy` ends the session's program and every process that the program left in its process group, then records the session as `destroyed`. diff --git a/src/batch/README.md b/src/batch/README.md index 821bf9b7..0c7481f7 100644 --- a/src/batch/README.md +++ b/src/batch/README.md @@ -32,7 +32,7 @@ This directory parses and runs a Batch: an ordered JSON list of `type`, `paste`, - A non-wait step that produced a seq, failed or not, becomes the `afterSeq` baseline for the next wait [executor.ts:executeBatch]. - A thrown error that is not a `CliError` becomes an `INTERNAL_ERROR` failed record [executor.ts:reframeAsInternalError]. - A waited `run` that does not complete fails with `WAIT_TIMEOUT` or `SESSION_NOT_RUNNING` [executor.ts:runRunStep]. -- A `wait` that times out or does not match fails with `WAIT_TIMEOUT` [executor.ts:runWaitStep]. When it timed out, the message names the step's timeout in ms. A `wait` whose RPC rejects (for example after repeated renderer failures) fails with that error's code [executor.ts:executeBatch]. +- A `wait` that times out or does not match fails with `WAIT_TIMEOUT` [executor.ts:runWaitStep]. When it timed out, the message names the step's timeout in ms. The host sends `eventLogStop` with a timed-out `wait` result. When it does, the message ends with `The event log stopped at seq (), so output after it is not seen.` Every failed `wait` step takes the stop from its own result, so a `--keep-going` batch never rereads the event log. A `wait` whose RPC rejects (for example after repeated renderer failures) fails with that error's code [executor.ts:executeBatch]. - All record schemas are `.strict()` [result.ts:BatchResultSchema]. - `buildPartialBatchResult` marks the first unrecorded step `interrupted` and the rest `not-run` [result.ts:buildPartialBatchResult]. - `createRpcStepDriver` asserts a non-empty `socketPath` and validates every RPC response with its result schema [stepDriver.ts:createRpcStepDriver]. diff --git a/src/batch/executor.ts b/src/batch/executor.ts index 5d6cf020..b53de142 100644 --- a/src/batch/executor.ts +++ b/src/batch/executor.ts @@ -8,6 +8,7 @@ import type { import type { StepDriver } from './stepDriver.js'; import { CliError } from '../cli/errors.js'; +import { eventLogStopTimeoutNote } from '../cli/eventLogStop.js'; import { ERROR_CODES, makeCliError } from '../protocol/errors.js'; import { unreachable } from '../util/assert.js'; import { unreachedStepRecord } from './result.js'; @@ -285,10 +286,15 @@ async function runWaitStep( ...observations, error: toBatchStepError( makeCliError(ERROR_CODES.WAIT_TIMEOUT, { - message: + message: [ result.timedOut && step.timeoutMs !== undefined ? `Render wait at step ${String(index)} timed out after ${String(step.timeoutMs)} ms.` : `Render wait at step ${String(index)} timed out before its condition was met.`, + // The host sends the stop with a timed-out result. + ...(result.eventLogStop === undefined + ? [] + : [eventLogStopTimeoutNote(result.eventLogStop)]), + ].join(' '), }), ), }; diff --git a/src/cli/README.md b/src/cli/README.md index 73993eeb..d1269650 100644 --- a/src/cli/README.md +++ b/src/cli/README.md @@ -14,6 +14,7 @@ This directory is the agent-tty command-line front end. `main.ts` is the bin ent - `emitSuccess`, `emitSuccessSync`, `emitFailure`, `emitCliError`, `argvHasFlag`, `exitAfterOutputFlush`, `formatSuccessOutput`, `writeJsonEnvelope`, `writeHumanLines`, `writeHumanErrorLines`, `writeStdoutSync`, `setColorEnabled`. [output.ts:emitSuccess] [output.ts:emitFailure] [output.ts:writeStdoutSync] [output.ts:argvHasFlag] [output.ts:exitAfterOutputFlush] - `resolveCommandTarget(options)` returns a `CommandTarget` (`sessionId`, `sessionDirectory`, `manifestPath`, `socketPath`, `manifest`). Also exports `ResolveCommandTargetOptions` and `CommandTargetManifest`. [commandTarget.ts:resolveCommandTarget] [commandTarget.ts:CommandTarget] - `assertSessionCommandable(manifest, sessionId)` and `checkedSessionDir(home, sessionId)`. [sessionGuards.ts:assertSessionCommandable] [sessionGuards.ts:checkedSessionDir] +- `describeEventLogStop(stop)` words where the stored log stopped, as `at seq (byte limit)`, `at seq (event limit)` or `at seq (reason unknown)`. `eventLogStopTimeoutNote(stop)` is the sentence that a `WAIT_TIMEOUT` message adds. The `unknown` reason reads `(reason unknown)`. `inspect`, `wait` and `batch` use them. None of them reads the event log for a wait: the stop comes with the host's wait result, or from the events of an offline replay. [eventLogStop.ts:describeEventLogStop] [eventLogStop.ts:eventLogStopTimeoutNote] - `exitCodeForError(code)`. [exitCodes.ts:exitCodeForError] - `warnIfUnsupportedNode(logger, version)`. [nodeRuntime.ts:warnIfUnsupportedNode] - `CliError` (`code`, `retryable`, `details`, `result`). `emitCliError` copies `result` into the error envelope when it is set. [errors.ts:CliError] [output.ts:emitCliError] diff --git a/src/cli/commands/README.md b/src/cli/commands/README.md index 92026ab8..5f5bddbe 100644 --- a/src/cli/commands/README.md +++ b/src/cli/commands/README.md @@ -32,7 +32,7 @@ This directory holds one handler module per agent-tty CLI command. `src/cli/main - `mouse` validates its options with `MouseParamsSchema` before it resolves the session, so a malformed `--row` or `--col` is `INVALID_INPUT` with command `mouse`. [mouse.ts:runMouseCommand] - Text input and `--file` are mutually exclusive. The file must be a regular file per `lstat` and at most `MAX_INPUT_FILE_SIZE` bytes. Violations throw `INVALID_INPUT`. [inputSource.ts:resolveCommandInputText] - `wait` rejects a mix of legacy flags (`--exit`, `--idle-ms`) and render flags. `--text` and `--regex` are mutually exclusive. The default timeout is `DEFAULT_WAIT_TIMEOUT_MS` (30,000 ms), which the `main.ts` help text names. An `--idle-ms` at or above the effective timeout (explicit or default), or a `--screen-stable-ms` whose `screenStableMinimumMs` is not shorter than it, throws `INVALID_DURATION`. `--timeout 0` skips the check. [wait.ts:runWaitCommand] [wait.ts:assertWindowFitsTimeout] [main.ts:main] -- A timed-out `wait` throws `WAIT_TIMEOUT` with the wait result attached, so its error envelope keeps `result`. Without `--json`, and only when the wait lasted its whole timeout (`timeoutMs > 0`, measured by the CLI), the message names the timeout and, for the default, how to change it. With `--json`, the message stays `Wait timed out before its condition was met.` [wait.ts:throwIfTimedOut] [wait.ts:timeoutMessage] +- A timed-out `wait` throws `WAIT_TIMEOUT` with the wait result attached, so its error envelope keeps `result`. Without `--json`, and only when the wait lasted its whole timeout (`timeoutMs > 0`, measured by the CLI), the message names the timeout and, for the default, how to change it. With `--json`, the message stays `Wait timed out before its condition was met.` The host sends `eventLogStop` with the result of a render wait that timed out. When it does, in both modes the message ends with `The event log stopped at seq (), so output after it is not seen.` and `details.eventLogStop` carries `{ reason, seq }`. The CLI never rereads the event log for a wait. `--idle-ms` and `--exit` waits carry no stop. When the host is unreachable, `wait` replays the stored log. A wait that the replayed screen does not satisfy fails with `REPLAY_ERROR`, and its message and `details.eventLogStop` carry the stop that `findEventLogStop` finds in the replayed events. [wait.ts:throwIfTimedOut] [wait.ts:timeoutMessage] [wait.ts:buildOfflineRenderWaitResult] - `snapshot` and `screenshot` try the live RPC only when the manifest status is `running`. They fall back to offline replay only on `HOST_UNREACHABLE` and rethrow other errors. [snapshot.ts:runSnapshotCommand] [screenshot.ts:runScreenshotCommand] - `record export` accepts only the formats in `RecordExportFormatSchema` (`asciicast`, `webm`, `svg`). `--animate` needs `svg`, and `svg` refuses `--timing`. An event log over the size limit fails with `EXPORT_ERROR` in every format, a message that names the log size and the limit, and `eventLogBytes` and `maxEventLogBytes` in `details`. [record-export.ts:runRecordExportCommand] - `batch` parses the whole plan with `parseBatchPlan` before it resolves the session target. [batch.ts:runBatchCommand] @@ -46,7 +46,7 @@ This directory holds one handler module per agent-tty CLI command. `src/cli/main - Session commands send RPCs over the session socket through `sendRpc`. The method names are `run`, `type`, `paste`, `sendKeys`, `mouse`, `mark`, `resize`, `signal`, `snapshot`, `screenshot`, `wait`, `waitForRender` and `inspect`. `create` also sends `inspect` while it polls. [run.ts:runRunCommand] [send-keys.ts:runSendKeysCommand] [wait.ts:runWaitCommand] [create.ts:runCreateCommand] - `create` calls `allocateSession` (`src/host/lifecycle.ts`) with a `launch` callback that records the session ID and starts the host process through `launchHost`. If `launchHost` or a later step of `allocateSession` throws, `create` removes the session directory with `fs.rm`. A bad `cwd`, shell path or command segment fails before `allocateSession` creates the directory, so no session directory is left behind. A failure between `mkdir` and the callback, such as a failed `chmod`, leaves the empty session directory in place. Any failure after `allocateSession` returned, a failed readiness poll included, writes the empty `create-failed` marker with `writeCreateFailedMarker` and logs a failed marker write at debug level (`writeMarkerOrLog`). On Windows, while that write fails, the manifest names no program and the host runs (`hostRuns`, where a probe that throws counts as an exited host), it writes the marker again every 100 ms, up to 50 times, and reads the manifest after each write. It then calls the host's `stop()` (on Windows only once the manifest names the program) and leaves the session directory in place. `create` rethrows the error with `details.sessionId` added, and an error that is not a `CliError` becomes `INTERNAL_ERROR`. In text mode it first writes `Session failed to start.` to stderr with `writeHumanErrorLines`. It registers the Home through `upsertHome`. [create.ts:runCreateCommand] [create.ts:writeMarkerOrLog] [create.ts:hostRuns] - `gc` reads session directories, calls `reconcileSession` (which can rewrite manifests), deletes collectable session directories with `fs.rm` outside dry-run mode, and deregisters Homes through `forgetHome`. [gc.ts:gcSessions] [gc.ts:runGcCommand] -- `inspect` calls `reconcileSession` when the host is unreachable, and for an ended session other than destroyed whose folder holds the destroy marker. When that repair throws, it shows the manifest as it reads. [inspect.ts:runInspectCommand] +- `inspect` calls `reconcileSession` when the host is unreachable, and for an ended session other than destroyed whose folder holds the destroy marker. When that repair throws, it shows the manifest as it reads. It reads `eventCount`, `eventLogBytes` and `eventLogStop` in one pass over `events.jsonl` with `scanEventLog`, for a live host and for offline replay alike, so `eventLogStop.seq` is never above `lastEventSeq`. It leaves `eventLogStop` out when the log holds no stop marker. Without `--json`, an `Event Log Stopped:` line follows `Event Log Bytes:`. [inspect.ts:runInspectCommand] [inspect.ts:formatSessionLines] [src/storage/eventLogCodec.ts:scanEventLog] - `destroy` and `list` call `destroySession` and `listSessions` from `src/host/lifecycle.ts`. [destroy.ts:runDestroyCommand] [list.ts:runListCommand] - `snapshot`, `screenshot`, `wait` and `record diff` run offline replay through `withOfflineReplayRenderer` (`src/replay/offlineReplay.ts`). `record diff` sends no RPC. [wait.ts:runOfflineRenderWait] [record-diff.ts:runRecordDiffCommand] - Offline `snapshot` and `screenshot` write artifact files into the session artifacts directory and append artifact-manifest entries through `publishArtifactWithRollback` (`src/storage/artifactManifest.ts`). `snapshot` writes a temp JSON file through `writeTextFileAtomic`, and `screenshot` gets a temp PNG from the renderer. `publishArtifactWithRollback` links or copies the temp file into place without replacing an existing file. [src/snapshot/capture.ts:captureSnapshotResult] [src/snapshot/capture.ts:persistSnapshotArtifact] [src/screenshot/capture.ts:captureScreenshotResult] diff --git a/src/cli/commands/inspect.ts b/src/cli/commands/inspect.ts index 4d94228d..cb8caada 100644 --- a/src/cli/commands/inspect.ts +++ b/src/cli/commands/inspect.ts @@ -10,11 +10,8 @@ import type { SessionRecord, SessionStatus } from '../../protocol/schemas.js'; import { CliError } from '../errors.js'; import type { CommandContext } from '../context.js'; +import { describeEventLogStop } from '../eventLogStop.js'; -import { - countEventLogEntries, - statEventLogBytes, -} from '../../host/eventLog.js'; import { isMarkedDestroyed, reconcileSession } from '../../host/lifecycle.js'; import { sendRpc } from '../../host/rpcClient.js'; import { @@ -26,6 +23,7 @@ import { deriveTerminationCategory } from '../../protocol/terminationCategory.js import { ERROR_CODES, makeCliError } from '../../protocol/errors.js'; import { emitSuccess } from '../output.js'; import { computeArtifactHealth } from '../../storage/artifactHealth.js'; +import { scanEventLog } from '../../storage/eventLogCodec.js'; import { readManifest, readManifestIfExists } from '../../storage/manifests.js'; import { eventLogPath, @@ -147,6 +145,12 @@ function formatSessionLines(result: InspectResult): string[] { lines.push(`Event Log Bytes: ${String(result.eventLogBytes)}`); } + if (result.eventLogStop !== undefined) { + lines.push( + `Event Log Stopped: ${describeEventLogStop(result.eventLogStop)}; later events are not recorded`, + ); + } + lines.push(`Uptime: ${String(uptime)}ms`); if (result.host !== undefined) { @@ -257,8 +261,12 @@ export async function runInspectCommand( } const eventLogFile = eventLogPath(sessionDirectory); - const eventCount = await countEventLogEntries(eventLogFile); - const eventLogBytes = await statEventLogBytes(eventLogFile); + // One read gives the count, the size and the stop, so a live host that + // writes the stop marker meanwhile cannot make them disagree. + const eventLogScan = await scanEventLog(eventLogFile); + const eventCount = eventLogScan?.entryCount ?? 0; + const eventLogBytes = eventLogScan?.byteCount; + const eventLogStop = eventLogScan?.stop; const uptime = computeUptime(session); let artifacts: ArtifactHealthSummary | undefined; try { @@ -294,6 +302,7 @@ export async function runInspectCommand( rendererRuntime, ...(host !== undefined ? { host } : {}), ...(eventLogBytes !== undefined ? { eventLogBytes } : {}), + ...(eventLogStop !== undefined ? { eventLogStop } : {}), }; emitSuccess({ diff --git a/src/cli/commands/wait.ts b/src/cli/commands/wait.ts index 7d196747..227adfc3 100644 --- a/src/cli/commands/wait.ts +++ b/src/cli/commands/wait.ts @@ -1,4 +1,5 @@ import type { + EventLogStop, WaitForRenderResult, WaitResult, } from '../../protocol/messages.js'; @@ -8,6 +9,7 @@ import type { SemanticSnapshot } from '../../renderer/types.js'; import { CliError } from '../errors.js'; import type { CommandContext } from '../context.js'; +import { eventLogStopTimeoutNote } from '../eventLogStop.js'; import { emitSuccess } from '../output.js'; import { sendRpc } from '../../host/rpcClient.js'; import { ERROR_CODES, makeCliError } from '../../protocol/errors.js'; @@ -27,6 +29,7 @@ import { import { computeScreenHash } from '../../renderer/canonicalScreen.js'; import { isTerminalSessionStatus } from '../../protocol/sessionStatusPolicy.js'; import { withOfflineReplayRenderer } from '../../replay/offlineReplay.js'; +import { findEventLogStop } from '../../storage/eventLogCodec.js'; import { readManifestIfExists } from '../../storage/manifests.js'; import { manifestPath, socketPath } from '../../storage/sessionPaths.js'; import { @@ -159,6 +162,9 @@ function timeoutMessage( return `Wait timed out after ${String(timeoutMs)} ms.`; } +// A render wait reads the event log, so output after the log stopped is not +// seen. The host sends `eventLogStop` with a timed-out result. The message +// says so, and `details` carries the same facts. function throwIfTimedOut( result: WaitResult | WaitForRenderResult, options: CommandOptions, @@ -166,17 +172,30 @@ function throwIfTimedOut( elapsedMs: number, ): void { if (result.timedOut) { + const message = timeoutMessage(options, timeoutMs, elapsedMs); + const eventLogStop = + 'eventLogStop' in result ? result.eventLogStop : undefined; throw makeCliError(ERROR_CODES.WAIT_TIMEOUT, { - message: timeoutMessage(options, timeoutMs, elapsedMs), - details: { sessionId: options.sessionId, timeoutMs }, + message: + eventLogStop === undefined + ? message + : `${message} ${eventLogStopTimeoutNote(eventLogStop)}`, + details: { + sessionId: options.sessionId, + timeoutMs, + ...(eventLogStop === undefined ? {} : { eventLogStop }), + }, result, }); } } +// `eventLogStop` comes from the events that the replay read, so the message +// names the stop that the snapshot was built from. function buildOfflineRenderWaitResult( preparedCondition: PreparedRenderWaitCondition, snapshot: SemanticSnapshot, + eventLogStop: EventLogStop | undefined, ): WaitForRenderResult { const match = matchRenderWaitSnapshot(preparedCondition, snapshot); @@ -194,9 +213,13 @@ function buildOfflineRenderWaitResult( }; } + const message = + 'Session host became unreachable during render wait, and the latest offline snapshot did not satisfy the requested wait condition.'; throw makeCliError(ERROR_CODES.REPLAY_ERROR, { message: - 'Session host became unreachable during render wait, and the latest offline snapshot did not satisfy the requested wait condition.', + eventLogStop === undefined + ? message + : `${message} ${eventLogStopTimeoutNote(eventLogStop)}`, details: { text: preparedCondition.text, regex: preparedCondition.regex, @@ -209,6 +232,7 @@ function buildOfflineRenderWaitResult( cursorRow: match.cursorRow, cursorCol: match.cursorCol, visibleLines: match.visibleLines, + ...(eventLogStop === undefined ? {} : { eventLogStop }), }, }); } @@ -220,9 +244,13 @@ async function runOfflineRenderWait( ): Promise { return await withOfflineReplayRenderer( { sessionDir: sessionDirectory, rendererName }, - async ({ backend }) => { + async ({ backend, replayInput }) => { const snapshot = await backend.snapshot({ includeScrollback: false }); - return buildOfflineRenderWaitResult(preparedCondition, snapshot); + return buildOfflineRenderWaitResult( + preparedCondition, + snapshot, + findEventLogStop(replayInput.events), + ); }, ); } diff --git a/src/cli/eventLogStop.ts b/src/cli/eventLogStop.ts new file mode 100644 index 00000000..c467269a --- /dev/null +++ b/src/cli/eventLogStop.ts @@ -0,0 +1,17 @@ +import type { EventLogStop, EventLogStopReason } from '../protocol/messages.js'; + +const STOP_REASON_TEXT: Record = { + bytes: 'byte limit', + events: 'event limit', + unknown: 'reason unknown', +}; + +/** Where the stored log stopped and why: `at seq 12 (byte limit)`. */ +export function describeEventLogStop(stop: EventLogStop): string { + return `at seq ${String(stop.seq)} (${STOP_REASON_TEXT[stop.reason]})`; +} + +/** The sentence that a `WAIT_TIMEOUT` message adds when the log has stopped. */ +export function eventLogStopTimeoutNote(stop: EventLogStop): string { + return `The event log stopped ${describeEventLogStop(stop)}, so output after it is not seen.`; +} diff --git a/src/host/README.md b/src/host/README.md index 31d4e292..ef4345cd 100644 --- a/src/host/README.md +++ b/src/host/README.md @@ -31,7 +31,7 @@ - `SessionState` holds the host's in-memory `SessionRecord` (type from `src/protocol/schemas.ts`). Methods: `snapshot`, `setChildPid`, `setDimensions`, `setFailureOrigin`, `requestDestroy`, `recordDestroyed` and `recordExit`. [sessionState.ts:SessionState] - `createSerialManifestWriter(path, snapshot, write?)` returns a function that writes `snapshot()` to `path`. The default `write` is `writeManifest` (`src/storage/manifests.ts`). `runHost` sends every manifest write through one such function. [manifestWriter.ts:createSerialManifestWriter] [hostMain.ts:runHost] - `EventLog` is the append-only JSONL event log with an in-memory buffer. Methods: `static open(filePath, options?)`, `append(type, payload)` (returns the `seq` of the stored record), `startAppend('input_run', payload)` (returns `{ seq, written }` without an `await`), `getEvents`, `getEventsSince(afterSeq)`, `readAll` and `close`. [eventLog.ts:EventLog] -- `MAX_EVENT_BUFFER_ENTRIES` is 250,000. `EVENT_LOG_TAIL_RESERVE_BYTES` is 4,096. `EVENT_LOG_TAIL_RESERVE_ENTRIES` is 2. `EVENT_LOG_LIMIT_LABEL_PREFIX` starts the label of the stop marker. `statEventLogBytes(filePath)` returns the file size or `undefined` when the file does not exist. `countEventLogEntries(filePath)` counts non-blank lines. [eventLog.ts:MAX_EVENT_BUFFER_ENTRIES] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_BYTES] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_ENTRIES] [eventLog.ts:EVENT_LOG_LIMIT_LABEL_PREFIX] [eventLog.ts:statEventLogBytes] [eventLog.ts:countEventLogEntries] +- `MAX_EVENT_BUFFER_ENTRIES` is 250,000. `EVENT_LOG_TAIL_RESERVE_BYTES` is 4,096. `EVENT_LOG_TAIL_RESERVE_ENTRIES` is 2. `EVENT_LOG_LIMIT_LABEL_PREFIX` (in `src/storage/eventLogCodec.ts`) starts the label of the stop marker. `EventLog.getStop()` returns `{ reason, seq }` for the stop marker in the records that `getEvents` returns, and `undefined` when there is none. The marker is in the buffer as soon as `recordStop` takes its `seq`, before its write settles, so `getStop` reports a stop that the file does not hold yet. The `waitForRender` handler returns it as `eventLogStop` with a timed-out result, so a wait never depends on what the file holds. `reason` is `bytes` or `events`, read from the label that `recordStop` builds with `eventLogStopLabel`. A marker with the reserved prefix whose label names no limit (a log that agent-tty did not write) is still a stop, with the reason `unknown`. [eventLog.ts:MAX_EVENT_BUFFER_ENTRIES] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_BYTES] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_ENTRIES] [eventLog.ts:EventLog.getStop] - `createMouseRpcHandler(deps)` returns the `mouse` handler `(request, rendererName)`, which resolves a `MouseResult` (`reported`, `bytesWritten`, `seq`). `MouseRpcDependencies` injects the renderer manager (in `runHost` a second `HostRendererManager` that only mouse input uses, so a render on another renderer or profile never replaces the encoder's backend and its same-cell motion state), the profile, the libghostty-vt probe, the replay loader, the commandable check, the PTY writer and the `input_mouse` append. The handler keeps the held buttons between requests. [mouseRpc.ts:createMouseRpcHandler] [mouseRpc.ts:MouseRpcDependencies] - `prepareHeldMouseButtons(currentHeld, input)` returns the next held set, `anyButtonPressed` and the `dragButton` that a button-free `move` reports, without changing `currentHeld`. [mouseInput.ts:prepareHeldMouseButtons] - `buildReplayInput(sessionId, manifest, events, ...)` builds a validated `ReplayInput` (type from `src/renderer/types.ts`) with the initial size and target sequence number. [replay.ts:buildReplayInput] @@ -44,7 +44,7 @@ ## Invariants -- `EventLog.open` derives the next `seq` from the last record (0 for an empty file), using `assertEventLogSize` and `parseEventLogContent` from `src/storage/eventLogCodec.ts`. `options.maxBytes` sets the size limit of the file. It defaults to `MAX_EVENT_LOG_SIZE`, must exceed `EVENT_LOG_TAIL_RESERVE_BYTES` and cannot exceed `MAX_EVENT_LOG_SIZE`, so the writer never passes what readers accept. `options.maxEntries` sets the limit of the number of events. It defaults to `MAX_EVENT_BUFFER_ENTRIES`, must exceed `EVENT_LOG_TAIL_RESERVE_ENTRIES` and cannot exceed `MAX_EVENT_BUFFER_ENTRIES`. Tests inject small limits. `append` asserts that the new `seq` equals the buffer length and validates the record with `EventRecordSchema`, so sequence numbers stay contiguous. A log that already holds its stop `marker` (a label that starts with `EVENT_LOG_LIMIT_LABEL_PREFIX`) starts stopped, so a reopened log stores only the `exit` record and writes no second stop record. A file of blank lines holds no record. When its size leaves less than `EVENT_LOG_TAIL_RESERVE_BYTES` free under `maxBytes`, `open` throws, because a stop record might not fit and a dropped append would have no earlier `seq` to report. [eventLog.ts:EventLog] [eventLog.ts:EventLogOptions] [eventLog.ts:EVENT_LOG_LIMIT_LABEL_PREFIX] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_BYTES] +- `EventLog.open` derives the next `seq` from the last record (0 for an empty file), using `assertEventLogSize` and `parseEventLogContent` from `src/storage/eventLogCodec.ts`. `options.maxBytes` sets the size limit of the file. It defaults to `MAX_EVENT_LOG_SIZE`, must exceed `EVENT_LOG_TAIL_RESERVE_BYTES` and cannot exceed `MAX_EVENT_LOG_SIZE`, so the writer never passes what readers accept. `options.maxEntries` sets the limit of the number of events. It defaults to `MAX_EVENT_BUFFER_ENTRIES`, must exceed `EVENT_LOG_TAIL_RESERVE_ENTRIES` and cannot exceed `MAX_EVENT_BUFFER_ENTRIES`. Tests inject small limits. `append` asserts that the new `seq` equals the buffer length and validates the record with `EventRecordSchema`, so sequence numbers stay contiguous. A log that already holds its stop `marker` (found by `findEventLogStop`) starts stopped, so a reopened log stores only the `exit` record and writes no second stop record. A file of blank lines holds no record. When its size leaves less than `EVENT_LOG_TAIL_RESERVE_BYTES` free under `maxBytes`, `open` throws, because a stop record might not fit and a dropped append would have no earlier `seq` to report. [eventLog.ts:EventLog] [eventLog.ts:EventLogOptions] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_BYTES] - At either limit, the file size or the number of events, `EventLog.append` stops the log. Ordinary records may fill the file up to `maxBytes` minus `EVENT_LOG_TAIL_RESERVE_BYTES`, and the buffer up to `maxEntries` minus `EVENT_LOG_TAIL_RESERVE_ENTRIES` (one entry for the stop record and one for the `exit` record). The first record that does not fit makes the log append one `marker` record with the label `agent-tty: event log limit reached (); later events are not recorded`, inside the reserves. `` is ` bytes` when the file size stopped the log and ` events` when the event count stopped it, so a reader can tell the two apart. A record that passes both limits is reported as the byte limit. From then on the log stores only the `exit` record, which also fits in the reserves, so the file never passes the byte limit and the buffer never passes the entry limit, and the host does not die from either. A record that the log does not store takes no `seq`. `append` resolves with the `seq` of the last stored record, and `startAppend` returns it as `seq`. Handlers, waited runs and `resize` therefore keep working, and the program keeps running. Its later output, input, resizes, signals and markers are in no snapshot, screenshot or export, because those replay the log. Every live feature that replays the log sees the program as it was at the stop: a live `wait` for text that appears after the stop times out, `wait --screen-stable-ms` can report a stable screen while the program keeps printing, and a mouse request after the stop encodes against the mouse modes that the stored log shows, not against modes that the program enabled or disabled after the stop (`reported` is false when the stored modes have mouse tracking off). `wait --idle-ms` and `wait --exit` follow the live program and are not affected. Once a write has failed, a dropped append rejects like any other. [eventLog.ts:EventLog] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_BYTES] [eventLog.ts:EVENT_LOG_TAIL_RESERVE_ENTRIES] [mouseRpc.ts:createMouseRpcHandler] - `EventLog.append` validates each payload with a strict Zod schema for its event type. It throws after `close()` and after an earlier write failed (the write queue stays rejected). It does not throw at a size limit or at `MAX_EVENT_BUFFER_ENTRIES`: the log stops instead. A failed write removes the failed records from the buffer. [eventLog.ts:EventLog] - `SessionState` enforces its transitions: `setChildPid` and `setDimensions` need status `running`, and the child PID is set once. `SessionState` has no method that sets `hostPid`. `requestDestroy` moves `running`, `exited` or `failed` to `destroying`. `recordDestroyed` needs `destroying`. `recordExit` needs `running` or `exiting` and no earlier exit. [sessionState.ts:SessionState] @@ -54,7 +54,7 @@ - These RPC handlers call `assertSessionCommandable` first: `type`, `mark`, `paste`, `run`, `sendKeys`, `mouse`, `resize`, `signal`, and `wait` in idle mode. [hostMain.ts:runHost] - The `mouse` handler throws `CAPABILITY_UNAVAILABLE` before it takes the renderer lock when the renderer is not `libghostty-vt` or the cached `probeLibghosttyVt` result has no mouse encoder. Inside the lock it replays until the backend holds the whole event log, then stages held buttons, encodes, writes to the PTY and calls `EventLog.append`, which takes the `seq` synchronously, with no `await` in between. So the encoder sees every mode change in the log, and the log orders mouse input as the PTY received it. Only a `RangeError` or `TypeError` from `encodeMouse`, such as a cell outside the grid, becomes `INVALID_INPUT`. [mouseRpc.ts:createMouseRpcHandler] - The `signal` handler accepts only `SIGTERM`, `SIGINT`, `SIGKILL`, `SIGHUP`, `SIGUSR1` and `SIGUSR2`, and throws `INVALID_SIGNAL` for others. [hostMain.ts:runHost] -- The `run` handler waits 30,000 ms by default. The `wait` handler needs exactly one of `exit` or `idleMs`, else it throws `INVALID_DURATION`. `waitForRender` polls every `RENDER_WAIT_POLL_INTERVAL_MS` (200 ms, `src/renderWait/limits.ts`). [hostMain.ts:runHost] +- The `run` handler waits 30,000 ms by default. The `wait` handler needs exactly one of `exit` or `idleMs`, else it throws `INVALID_DURATION`. `waitForRender` polls every `RENDER_WAIT_POLL_INTERVAL_MS` (200 ms, `src/renderWait/limits.ts`). Its timeout result carries `eventLogStop` from `EventLog.getStop()` when the log has stopped. [hostMain.ts:runHost] - `runHost` handles PTY exit once. If the status is `destroying` it records `destroyed`, else `exited`. Shutdown runs once. [hostMain.ts:runHost] - `destroySession` throws `SESSION_NOT_FOUND` when there is no manifest, or when the session directory is gone at its `destroyed` write, and `SESSION_ALREADY_DESTROYED` for a destroyed session. When it records `destroyed` itself, `recordDestroyed` first writes the empty `/destroyed` marker with `writeDestroyMarker`. Before it records a terminal session as `destroyed`, it calls `stopProgramGroup` with the group from `readProvenProgramGroup`, so only a group with a token gets a signal, and it throws `HOST_TIMEOUT` while that group lives. A session already recorded `destroyed` with the unshared group gets no wait, because only a process that saw the group recorded it, after its wait. A running session gets the `destroy` RPC. With `force` it gets `SIGKILL` for the program's group and, only when `hostMatches` proves the host, for the host: on Linux `judgeHost` must return `alive`, elsewhere `/bin/ps` must show `_host `, and on Windows the host gets no signal. It polls every 100 ms for at most 50 attempts and throws `HOST_TIMEOUT` if the status is still not terminal. With `force`, when the read of `program.json` or the first `isProgramGroupAlive` of the group throws an error that `isUnreadableGroupError` accepts (a `STORAGE_READ_ERROR` whose cause, or a `/proc` read whose code, is in `UNREADABLE_ERRNOS`), `stopUnlessUnreadable` waits `DESTROY_POLL_INTERVAL_MS` (100 ms, abortable) and reads and judges the group again. Only when that read throws such an error too does it stop and wait for the host alone, and `destroySession` records `destroyed` once the host has ended. An error after that first judgement, as in the wait, still throws. `forceSkipsGroup` runs that read and first judgement and returns whether such an error came, so `gcSessions` names `destroy --force` for a failed reconcile only then. [lifecycle.ts:destroySession] [programGroup.ts:isUnreadableGroupError] [programGroup.ts:forceSkipsGroup] - `reconcileSession` writes `destroyed` (both PIDs `null`, the exit code kept, no failure fields) over a terminal manifest other than `destroyed` when the destroy marker exists (`writeMarkedDestroyed`). It then unlinks the socket, removes the Home's socket directory if it is empty, removes the old socket at `legacySocketPath` when `isOwnLegacySocket` accepts it, and returns. Otherwise it does nothing when the manifest is missing, the status is terminal or the host can be alive: `judgeHost` with the record from `host.json` does not return `dead`. A null `hostPid` counts as a dead host. `allocateSession` writes the host PID into the first manifest, so a new session already has a `hostPid`. After the liveness check, `reconcileSession` reads the manifest again and does nothing when it is now missing or terminal. Otherwise it maps the status it read again: `running` to `failed` (failure origin `host-death`), `exiting` to `exited`, `destroying` to `destroyed` and any other status to `failed` (origin `unknown`), and sets both PIDs to `null`. Right after that write, when its status was not `destroyed` and the destroy marker exists, it writes `destroyed` again from the same copy, with both PIDs `null` and no failure fields. When the session directory is gone at any of these writes, it returns. [lifecycle.ts:reconcileSession] [lifecycle.ts:writeMarkedDestroyed] diff --git a/src/host/eventLog.ts b/src/host/eventLog.ts index 96482fdc..07e1e2b1 100644 --- a/src/host/eventLog.ts +++ b/src/host/eventLog.ts @@ -1,7 +1,5 @@ -import { createReadStream } from 'node:fs'; -import { open, readFile, stat } from 'node:fs/promises'; +import { open, readFile } from 'node:fs/promises'; import type { FileHandle } from 'node:fs/promises'; -import { createInterface } from 'node:readline'; import { z } from 'zod'; @@ -11,6 +9,7 @@ import { InputRunEventPayloadSchema, MarkerEventPayloadSchema, RunCompleteEventPayloadSchema, + type EventLogStop, type EventRecord, type InputMouseEventPayload, type InputRunEventPayload, @@ -19,12 +18,15 @@ import { } from '../protocol/schemas.js'; import { assertEventLogSize, + EVENT_LOG_LIMIT_LABEL_PREFIX, + eventLogStopLabel, + findEventLogStop, MAX_EVENT_LOG_SIZE, parseEventLogContent, + type EventLogLimit, } from '../storage/eventLogCodec.js'; import { ERROR_CODES, makeCliError } from '../protocol/errors.js'; import { invariant } from '../util/assert.js'; -import { hasErrorCode } from '../util/hasErrorCode.js'; const OutputEventPayloadSchema = z .object({ @@ -134,14 +136,6 @@ export const EVENT_LOG_TAIL_RESERVE_BYTES = 4096; */ export const EVENT_LOG_TAIL_RESERVE_ENTRIES = 2; -/** - * Start of the label of the marker that records why the log stopped. The rest - * of the label names the limit that was reached: `( bytes)` for the file - * size, `( events)` for the entry count. - */ -export const EVENT_LOG_LIMIT_LABEL_PREFIX = - 'agent-tty: event log limit reached'; - export interface EventLogOptions { /** * Size limit of the file in bytes. Defaults to `MAX_EVENT_LOG_SIZE`, which @@ -155,9 +149,6 @@ export interface EventLogOptions { readonly maxEntries?: number; } -/** The limit that stopped the log. */ -type StopReason = 'bytes' | 'entries'; - interface PreparedRecord { readonly record: EventRecord; readonly line: string; @@ -243,62 +234,6 @@ function deriveNextSeq(records: readonly EventRecord[]): number { return lastRecord.seq + 1; } -/** True for the marker that `recordStop` writes when the log stops. */ -function isStopMarker(record: EventRecord): boolean { - return ( - record.type === 'marker' && - record.payload.label.startsWith(EVENT_LOG_LIMIT_LABEL_PREFIX) - ); -} - -/** - * Returns the event log file size in bytes, or `undefined` when the file - * does not exist (ENOENT). Non-ENOENT errors propagate to the caller. - */ -export async function statEventLogBytes( - filePath: string, -): Promise { - assertFilePath(filePath); - - try { - const stats = await stat(filePath); - return stats.size; - } catch (error: unknown) { - if (hasErrorCode(error, 'ENOENT')) { - return undefined; - } - - throw error; - } -} - -export async function countEventLogEntries(filePath: string): Promise { - assertFilePath(filePath); - - let count = 0; - - try { - const lineReader = createInterface({ - input: createReadStream(filePath, { encoding: 'utf8' }), - crlfDelay: Infinity, - }); - - for await (const line of lineReader) { - if (line.trim().length > 0) { - count += 1; - } - } - } catch (error: unknown) { - if (hasErrorCode(error, 'ENOENT')) { - return 0; - } - - throw error; - } - - return count; -} - export class EventLog { // The writeQueue is intentionally poisoned on rejection. Once any append // fails, every subsequent queued write inherits that rejection so that @@ -403,7 +338,7 @@ export class EventLog { maxBytes, maxEntries, eventBuffer, - eventBuffer.some(isStopMarker), + findEventLogStop(eventBuffer) !== undefined, ); } catch (error) { await fileHandle.close(); @@ -445,7 +380,7 @@ export class EventLog { payload: EventLogPayload, ): Promise { // Only `recordStop` writes the stop marker, so a reopened log can trust a - // marker with this prefix (see `isStopMarker`). + // marker with this prefix (see `findEventLogStop`). if ( type === 'marker' && (payload as MarkerEventPayload).label.startsWith( @@ -540,7 +475,7 @@ export class EventLog { private findStopReason( type: EventLogEventType, bytes: number, - ): StopReason | undefined { + ): EventLogLimit | undefined { const reserveBytes = type === 'exit' ? 0 : EVENT_LOG_TAIL_RESERVE_BYTES; const reserveEntries = type === 'exit' ? 0 : EVENT_LOG_TAIL_RESERVE_ENTRIES; @@ -548,7 +483,7 @@ export class EventLog { return 'bytes'; } if (this.eventBuffer.length + 1 > this.maxEntries - reserveEntries) { - return 'entries'; + return 'events'; } return undefined; } @@ -560,15 +495,14 @@ export class EventLog { * reached: bytes or events. It goes into the reserves, so it fits unless a * log that this host did not write already filled them before it opened. */ - private recordStop(reason: StopReason): void { + private recordStop(reason: EventLogLimit): void { this.stopped = true; - const limit = - reason === 'bytes' - ? `${String(this.maxBytes)} bytes` - : `${String(this.maxEntries)} events`; const stop = this.prepareRecord('marker', { - label: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (${limit}); later events are not recorded`, + label: eventLogStopLabel( + reason, + reason === 'bytes' ? this.maxBytes : this.maxEntries, + ), }); if ( stop.bytes > this.maxBytes - this.fileBytes || @@ -658,6 +592,17 @@ export class EventLog { return this.eventBuffer; } + /** + * Where this log stopped, from the records that `getEvents` returns: the + * stop marker is there as soon as `recordStop` takes its `seq`, before its + * write settles. So a caller that reads the stop here sees every stop that + * its own reads of the buffer can see. The file on disk can still lack the + * marker at that moment. + */ + getStop(): EventLogStop | undefined { + return findEventLogStop(this.eventBuffer); + } + getEventsSince(afterSeq: number): EventRecord[] { invariant(Number.isInteger(afterSeq), 'afterSeq must be an integer'); invariant(afterSeq >= -1, 'afterSeq must be greater than or equal to -1'); diff --git a/src/host/hostMain.ts b/src/host/hostMain.ts index f01f07c9..e1b5857d 100644 --- a/src/host/hostMain.ts +++ b/src/host/hostMain.ts @@ -1200,10 +1200,14 @@ export async function runHost( // Best-effort snapshot for timeout reporting. } + // The host's own log tells whether it stopped, so the wait + // does not depend on what the file holds yet. + const eventLogStop = eventLog.getStop(); return { matched: false, timedOut: true, capturedAtSeq: latestCapturedAtSeq, + ...(eventLogStop === undefined ? {} : { eventLogStop }), }; }, }), diff --git a/src/protocol/README.md b/src/protocol/README.md index 38d8becf..74476e42 100644 --- a/src/protocol/README.md +++ b/src/protocol/README.md @@ -16,9 +16,9 @@ - `SessionStatusPolicy` and `getSessionStatusPolicy(status)` give the per-status flags `active`, `terminal`, `commandable`, `liveHostEligible`, `offlineReplayEligible`, `collectable` and `destroyed`. The predicates `isActiveSessionStatus`, `isTerminalSessionStatus`, `isCommandableSessionStatus`, `isLiveHostEligibleSessionStatus`, `isOfflineReplayEligibleSessionStatus`, `isCollectableSessionStatus` and `isDestroyedSessionStatus` read the same table. [sessionStatusPolicy.ts:getSessionStatusPolicy] [sessionStatusPolicy.ts:isCommandableSessionStatus] - Session manifest schemas: `SessionStatusSchema` (`running`, `exiting`, `exited`, `failed`, `destroying`, `destroyed`), `FailureOriginSchema` (`host-death`, `renderer-failure`, `storage-corruption`, `unknown`), `SessionRecordSchema` and `Sha256HexSchema`, with their inferred types. [schemas.ts:SessionStatusSchema] [schemas.ts:FailureOriginSchema] [schemas.ts:SessionRecordSchema] [schemas.ts:Sha256HexSchema] - Event log schemas: `EventTypeSchema` (`output`, `input_text`, `input_paste`, `input_keys`, `input_mouse`, `input_run`, `run_complete`, `resize`, `signal`, `exit`, `marker`), one `*EventPayloadSchema` and one `*EventRecordSchema` per type, and `EventRecordSchema` with type `EventRecord`, a union discriminated on `type`. [schemas.ts:EventTypeSchema] [schemas.ts:EventRecordSchema] -- Snapshot, screenshot, wait and record schemas: `SnapshotParamsSchema`, `SnapshotResultSchema` (discriminated on `format`: `structured` or `text`), `ScreenshotParamsSchema`, `ScreenshotResultSchema` (with optional `missingGlyphs` entries from `MissingGlyphSchema`: a `U+XXXX` `codePoint` and at least one `{ row, col }` cell, and an optional `missingGlyphsTruncated: true`), `RenderWaitScopeSchema`, `WaitForRenderParamsSchema`, `WaitForRenderResultSchema`, `WaitResultSchema`, `RecordExportResultSchema` (discriminated on `format`: only `webm` allows optional `missingGlyphs` entries from `RecordedMissingGlyphSchema`, a `U+XXXX` `codePoint` and no cells, and an optional `missingGlyphsTruncated: true`), `RecordDiffResultSchema` and `ReplayTimingModeSchema` (`recorded`, `accelerated`, `max-speed`), with related schemas and types. [schemas.ts:SnapshotResultSchema] [schemas.ts:WaitForRenderParamsSchema] [schemas.ts:RecordDiffResultSchema] [schemas.ts:ReplayTimingModeSchema] +- Snapshot, screenshot, wait and record schemas: `SnapshotParamsSchema`, `SnapshotResultSchema` (discriminated on `format`: `structured` or `text`), `ScreenshotParamsSchema`, `ScreenshotResultSchema` (with optional `missingGlyphs` entries from `MissingGlyphSchema`: a `U+XXXX` `codePoint` and at least one `{ row, col }` cell, and an optional `missingGlyphsTruncated: true`), `RenderWaitScopeSchema`, `WaitForRenderParamsSchema`, `WaitForRenderResultSchema` (a timed-out result has an optional `eventLogStop`), `EventLogStopSchema` (`{ reason, seq }`, strict) with `EventLogStopReasonSchema` (`bytes`, `events` or `unknown`), `WaitResultSchema`, `RecordExportResultSchema` (discriminated on `format`: only `webm` allows optional `missingGlyphs` entries from `RecordedMissingGlyphSchema`, a `U+XXXX` `codePoint` and no cells, and an optional `missingGlyphsTruncated: true`), `RecordDiffResultSchema` and `ReplayTimingModeSchema` (`recorded`, `accelerated`, `max-speed`), with related schemas and types. [schemas.ts:SnapshotResultSchema] [schemas.ts:WaitForRenderParamsSchema] [schemas.ts:RecordDiffResultSchema] [schemas.ts:ReplayTimingModeSchema] - `schemas.ts` and `messages.ts` re-export the capability and renderer-runtime schemas (`CapabilityEntrySchema`, `RendererRuntimeSummarySchema` and others) from `src/renderer/capabilities.ts`. [src/renderer/capabilities.ts:CapabilityEntrySchema] -- `messages.ts` also re-exports these `schemas.ts` schemas: `MissingGlyphSchema`, `RecordedMissingGlyphSchema`, `RecordDiffLineSchema`, `RecordDiffResultSchema`, `RecordDiffSideSchema`, `RecordExportResultSchema`, `ReplayTimingModeSchema`, `RichSnapshotLineSchema`, `SnapshotCellSchema`, `SnapshotParamsSchema`, `SnapshotResultSchema`, `ScreenshotParamsSchema`, `ScreenshotResultSchema`, `WaitForRenderParamsSchema`, `WaitForRenderResultSchema` and `WaitResultSchema`. [schemas.ts:RecordDiffResultSchema] [schemas.ts:WaitResultSchema] +- `messages.ts` also re-exports these `schemas.ts` schemas: `MissingGlyphSchema`, `RecordedMissingGlyphSchema`, `RecordDiffLineSchema`, `RecordDiffResultSchema`, `RecordDiffSideSchema`, `RecordExportResultSchema`, `ReplayTimingModeSchema`, `RichSnapshotLineSchema`, `SnapshotCellSchema`, `SnapshotParamsSchema`, `SnapshotResultSchema`, `ScreenshotParamsSchema`, `ScreenshotResultSchema`, `WaitForRenderParamsSchema`, `WaitForRenderResultSchema`, `WaitResultSchema`, `EventLogStopSchema` and `EventLogStopReasonSchema`. [schemas.ts:RecordDiffResultSchema] [schemas.ts:WaitResultSchema] - RPC envelope schemas: `RpcRequestSchema`, `RpcErrorSchema`, `RpcSuccessResponseSchema`, `RpcErrorResponseSchema` and `RpcResponseSchema` (discriminated on `ok`), with their types. [messages.ts:RpcRequestSchema] [messages.ts:RpcResponseSchema] - `RpcMethodSchema` and `RpcMethod` list the host RPC methods: `inspect`, `snapshot`, `screenshot`, `type`, `paste`, `run`, `mark`, `sendKeys`, `mouse`, `resize`, `signal`, `wait`, `waitForRender` and `destroy`. `RpcMethodSchemas` maps each method to its `{ params, result }` schemas. [messages.ts:RpcMethodSchema] [messages.ts:RpcMethodSchemas] - Per-method schemas: `InspectParamsSchema`, `HostInspectResultSchema`, `HostInfoSchema`, `InspectResultSchema`, `ArtifactHealthSummarySchema`, `TerminationCategorySchema`, and the `Type`, `Paste`, `Run`, `Mark`, `SendKeys`, `Mouse`, `Resize`, `Signal` and `Destroy` params and result schemas, plus `WaitParamsSchema`. [messages.ts:InspectResultSchema] [messages.ts:TerminationCategorySchema] [messages.ts:RunParamsSchema] diff --git a/src/protocol/messages.ts b/src/protocol/messages.ts index 38accc9e..025a99ef 100644 --- a/src/protocol/messages.ts +++ b/src/protocol/messages.ts @@ -1,6 +1,8 @@ import { z } from 'zod'; import type { + EventLogStop as EventLogStopType, + EventLogStopReason as EventLogStopReasonType, MissingGlyph as MissingGlyphType, RecordedMissingGlyph as RecordedMissingGlyphType, RecordDiffLine as RecordDiffLineType, @@ -13,6 +15,7 @@ import type { } from './schemas.js'; import { + EventLogStopSchema, RendererRuntimeSummarySchema, ScreenshotParamsSchema, ScreenshotResultSchema, @@ -28,6 +31,8 @@ import { import { RendererNameSchema } from '../renderer/names.js'; export { + EventLogStopReasonSchema, + EventLogStopSchema, MissingGlyphSchema, RecordedMissingGlyphSchema, RecordDiffLineSchema, @@ -215,14 +220,25 @@ export const InspectResultSchema = z // mode (`usedOfflineReplay === true` or the session is not live-host // eligible). host: HostInfoSchema.optional(), - // Populated in both live and offline-replay modes from a `stat()` on the - // session's `events.jsonl`. Absent only when the event log file is - // missing on disk (e.g. a session that crashed before its first write). + // Populated in both live and offline-replay modes: the bytes that one + // read of the session's `events.jsonl` returned, the same read that + // counts `eventCount` and finds `eventLogStop`. Absent only when the + // event log file is missing on disk (e.g. a session that crashed before + // its first write). eventLogBytes: z.number().int().nonnegative().optional(), + // Present only when the stored event log holds the stop marker, so + // later events (and so later output) are missing from every snapshot, + // wait and export. Read from `events.jsonl` in one pass with + // `eventCount`, in live and offline modes. + eventLogStop: EventLogStopSchema.optional(), }) .strict(); export type InspectResult = z.infer; +export type EventLogStopReason = EventLogStopReasonType; + +export type EventLogStop = EventLogStopType; + export type ReplayTimingMode = ReplayTimingModeType; export type SnapshotCell = SnapshotCellType; diff --git a/src/protocol/schemas.ts b/src/protocol/schemas.ts index 526a1f33..80baa74c 100644 --- a/src/protocol/schemas.ts +++ b/src/protocol/schemas.ts @@ -562,6 +562,23 @@ export const WaitResultSchema = z .strict(); export type WaitResult = z.infer; +// Why the event log stopped taking records (see `EventLog.queueAppend`): +// `bytes` when the file reached its size limit, `events` when it reached its +// entry limit, `unknown` when the stop marker is in the log but its label +// names no limit (a log that agent-tty did not write). +export const EventLogStopReasonSchema = z.enum(['bytes', 'events', 'unknown']); +export type EventLogStopReason = z.infer; + +// Where the event log stopped: the `seq` of the stop marker that the host +// writes into `events.jsonl`, and the limit it names. +export const EventLogStopSchema = z + .object({ + reason: EventLogStopReasonSchema, + seq: NonNegativeIntSchema, + }) + .strict(); +export type EventLogStop = z.infer; + export const WaitForRenderResultSchema = z .object({ matched: z.boolean(), @@ -571,6 +588,10 @@ export const WaitForRenderResultSchema = z cursorCol: NonNegativeIntSchema.optional(), capturedAtSeq: ReplaySeqSchema, screenHash: Sha256HexSchema.optional(), + // Only on a timed-out wait, and only when the host's event log has + // stopped. A wait reads the event log, so output after this seq is not + // seen. + eventLogStop: EventLogStopSchema.optional(), }) .strict(); const RecordExportResultFields = { diff --git a/src/storage/README.md b/src/storage/README.md index 8e462fed..cedcd2c9 100644 --- a/src/storage/README.md +++ b/src/storage/README.md @@ -18,7 +18,7 @@ - `readValidatedJsonFile`, `writeValidatedJsonFile`, `writeValidatedJsonFileSync` and `writeTextFileAtomic`, with their option interfaces, are the generic validated JSON and text file helpers. [manifests.ts:readValidatedJsonFile] [manifests.ts:writeValidatedJsonFile] [manifests.ts:writeValidatedJsonFileSync] [manifests.ts:writeTextFileAtomic] - `readManifest`, `readManifestIfExists` and `writeManifest` read and write a `SessionRecord` (schema from `src/protocol/schemas.ts`). `readManifestIfExists` returns `null` when the file does not exist. [manifests.ts:readManifest] [manifests.ts:readManifestIfExists] [manifests.ts:writeManifest] - `MAX_EVENT_LOG_SIZE` is 52,428,800 bytes (50 MB). `assertEventLogSize(size)` checks a size against it and throws `EventLogTooLargeError` (with `size` and `maxSize`) above it. The host writer stops at this size (see `src/host/README.md`), so only a log from an older host can exceed it. [eventLogCodec.ts:MAX_EVENT_LOG_SIZE] [eventLogCodec.ts:assertEventLogSize] [eventLogCodec.ts:EventLogTooLargeError] -- `parseEventLogContent(text)` parses JSONL text into `EventRecord[]`. `validateEventRecords(values)` validates records that are already loaded. `readEventLogRecords(filePath)` checks the file size, reads the file and parses it. `EventRecordSchema` comes from `src/protocol/schemas.ts`. [eventLogCodec.ts:parseEventLogContent] [eventLogCodec.ts:validateEventRecords] [eventLogCodec.ts:readEventLogRecords] +- `parseEventLogContent(text)` parses JSONL text into `EventRecord[]`. `validateEventRecords(values)` validates records that are already loaded. `readEventLogRecords(filePath)` checks the file size, reads the file and parses it. `EventRecordSchema` comes from `src/protocol/schemas.ts`. `EVENT_LOG_LIMIT_LABEL_PREFIX` starts the label of the stop marker, `eventLogStopLabel(reason, limit)` builds that label, and `findEventLogStop(records)` returns `{ reason, seq }` for the first stop marker in `records` (`reason` is `bytes`, `events` or `unknown`, and `unknown` means that the label names no limit), or `undefined`. `scanEventLog(filePath)` streams the file once and returns `{ entryCount, byteCount, stop }` for one snapshot of it, or `undefined` when the file does not exist. It does not enforce `MAX_EVENT_LOG_SIZE`, and it looks for the stop only in the lines that `parseEventLogContent` would accept, up to the first bad line. `inspect` uses it. [eventLogCodec.ts:parseEventLogContent] [eventLogCodec.ts:validateEventRecords] [eventLogCodec.ts:readEventLogRecords] [eventLogCodec.ts:findEventLogStop] [eventLogCodec.ts:scanEventLog] - `ArtifactEntrySchema` and `ArtifactManifestSchema`, with types `ArtifactEntry` and `ArtifactManifest`, describe `/artifacts/manifest.json`. [artifactManifest.ts:ArtifactEntrySchema] [artifactManifest.ts:ArtifactManifestSchema] - `readArtifactManifest(sessionDir)` returns the manifest, or an empty version 1 manifest when the file does not exist. `writeArtifactManifest` validates and writes it. [artifactManifest.ts:readArtifactManifest] [artifactManifest.ts:writeArtifactManifest] - `appendArtifactWithRollback(options)` takes `AppendArtifactWithRollbackOptions` (`sessionDir`, `entry`, optional `rollbackArtifactPath` and `outputFileIdentity`) and adds one entry to the manifest. `outputFileIdentity` is the `fileIdentity` (device, inode, size and modification time in nanoseconds, all exact as bigints) of the entry's file, taken when it was written. With it, the append first drops the entries whose `metadata.outputPath` names that file, because the new file replaced theirs. It checks only the entries whose path differs from the new entry's at most in case (after Unicode NFC normalization), so it reads no other path. When the entry's `metadata.outputPath` no longer names that file (a later export replaced it, or a file server or a tool changed its modification time), the append drops nothing and still adds the entry, as before this option existed. When the path holds a file of another size, a later write replaced the entry's bytes, and the append adds nothing. The entry must then have an absolute `metadata.outputPath`. If the append fails and `rollbackArtifactPath` is set, it tries to delete that file, unless the manifest already lists a file of that name. It ignores `rm` errors. `createArtifactEntry` adds a ULID `id` and a `createdAt` time. [artifactManifest.ts:appendArtifactWithRollback] [artifactManifest.ts:AppendArtifactWithRollbackOptions] [artifactManifest.ts:createArtifactEntry] @@ -51,7 +51,7 @@ - `resolveHome` reads `process.env.AGENT_TTY_HOME` and calls `os.homedir()` and `realpathSync`. [home.ts:resolveHome] - `ensureHome` calls `mkdir(home, { recursive: true })`. It runs `chmod(home, 0o700)` only when `mkdir` created the directory. [home.ts:ensureHome] - `writeTextFileAtomic` creates the parent directory unless `createParentFolders` is `false`, writes `.tmp-` with mode `0o600`, then renames it to ``. `beforeRename` runs between the write and the rename with the temp file's path, and a `CliError` it throws is passed on unwrapped. On failure it removes the temp file. It does not call `fsync`. `writeValidatedJsonFile`, `writeArtifactManifest` and the registry writes all go through it. `writeManifest` writes and renames the temp file the same way but creates no directory, so it cannot bring back a session directory that `gc` removed. `writeValidatedJsonFileSync` does the same with `writeFileSync` and `renameSync` and creates no directory, and its type makes the caller pass `createParentFolders: false`. [manifests.ts:writeTextFileAtomic] [manifests.ts:writeManifest] [manifests.ts:writeValidatedJsonFileSync] -- `readValidatedJsonFile` and `readEventLogRecords` read files with `fs.promises.readFile`. `readEventLogRecords` also calls `stat`. [manifests.ts:readValidatedJsonFile] [eventLogCodec.ts:readEventLogRecords] +- `readValidatedJsonFile` and `readEventLogRecords` read files with `fs.promises.readFile`. `readEventLogRecords` also calls `stat`. `scanEventLog` opens the file, calls `stat` on the handle and reads a stream of the first `size` bytes. [manifests.ts:readValidatedJsonFile] [eventLogCodec.ts:readEventLogRecords] [eventLogCodec.ts:scanEventLog] - `ensureArtifactsDir` runs `mkdir(/artifacts)` without `recursive`, so it never creates the session directory, and it throws `SESSION_NOT_FOUND` when that directory is gone. When the artifacts path exists but is not a folder, or is a symlink whose target is missing, it throws `STORAGE_WRITE_ERROR`. When the session folder is gone by the time it checks that path, it throws `SESSION_NOT_FOUND`. Any other `stat` error (for example `ELOOP` from a symlink loop) gives `STORAGE_WRITE_ERROR` "Failed to check artifacts path … ()." with `details.errno` and the `stat` error as its cause. `ensureArtifactsDir` calls `stat` on the artifacts path when `mkdir` reports `EEXIST`. `sessionNotFoundIfRemoved` calls `stat` on the session directory. The other path functions are pure, and `socketPath` only hashes strings with `crypto`. [artifactPaths.ts:ensureArtifactsDir] [artifactPaths.ts:sessionNotFoundIfRemoved] [sessionPaths.ts:socketPath] - `publishArtifactFile` calls `link(temporaryPath, )` for each candidate name. If `link` fails with `EPERM`, `ENOTSUP`, `EOPNOTSUPP` or `ENOSYS` (a filesystem without hard links), it calls `copyFile` with `COPYFILE_EXCL` instead, so a reader can see a partly copied file until the copy finishes. When a name is taken, it compares bytes with `readFile` on the temp file and on that file, which it opens with `O_NOFOLLOW` and `O_NONBLOCK` and rejects unless it is a regular file, so a symlink or a FIFO at the name fails at once. If that file is gone by then, it retries the same name. One call allows 3 such retries in total, shared across all names. It always removes the temp file with `rm({ force: true })`. [artifactFiles.ts:publishArtifactFile] - `appendArtifactWithRollback` reads, changes and writes the artifact manifest. On failure it calls `rm(rollbackArtifactPath, { force: true })`, unless the manifest already lists a file of that name. With `outputFileIdentity`, it calls `stat` on the entry's `metadata.outputPath` and on those listed paths that differ from it at most in case. [artifactManifest.ts:appendArtifactWithRollback] diff --git a/src/storage/eventLogCodec.ts b/src/storage/eventLogCodec.ts index 174bd1a8..970c9e0e 100644 --- a/src/storage/eventLogCodec.ts +++ b/src/storage/eventLogCodec.ts @@ -1,10 +1,40 @@ -import { readFile, stat } from 'node:fs/promises'; +import { open, readFile, stat } from 'node:fs/promises'; +import type { FileHandle } from 'node:fs/promises'; -import { EventRecordSchema, type EventRecord } from '../protocol/schemas.js'; +import { + EventRecordSchema, + type EventLogStop, + type EventLogStopReason, + type EventRecord, +} from '../protocol/schemas.js'; import { invariant } from '../util/assert.js'; +import { hasErrorCode } from '../util/hasErrorCode.js'; export const MAX_EVENT_LOG_SIZE = 50 * 1024 * 1024; +/** + * Start of the label of the marker that records why the log stopped. The rest + * of the label names the limit that was reached: `( bytes)` for the file + * size, `( events)` for the entry count. The host writer reserves the + * prefix, so a marker that carries it is the stop marker. + */ +export const EVENT_LOG_LIMIT_LABEL_PREFIX = + 'agent-tty: event log limit reached'; + +/** The limits that stop a log. A reader may also find `unknown`. */ +export type EventLogLimit = Exclude; + +/** + * The label of the stop marker. It names the limit that stopped the log, as + * `( bytes)` or `( events)`, and `findEventLogStop` reads it back. + */ +export function eventLogStopLabel( + reason: EventLogLimit, + limit: number, +): string { + return `${EVENT_LOG_LIMIT_LABEL_PREFIX} (${String(limit)} ${reason}); later events are not recorded`; +} + /** * A log file larger than the limit. This is a state of the file, not a bug in * the caller, so commands can catch it and name the cause. The host writer @@ -49,23 +79,23 @@ function parseEventLogLine(line: string, lineNumber: number): EventRecord { return parsedRecord.data; } -function assertContiguousSequence(records: readonly EventRecord[]): void { - if (records.length === 0) { +function assertSeqFollows(previousSeq: number | undefined, seq: number): void { + if (previousSeq === undefined) { + invariant(seq === 0, 'first event log seq must be 0'); return; } - invariant(records[0]?.seq === 0, 'first event log seq must be 0'); - - for (let index = 1; index < records.length; index += 1) { - const previous = records[index - 1]; - const current = records[index]; + invariant( + seq === previousSeq + 1, + 'event log seq values must increase by 1 without gaps', + ); +} - invariant(previous !== undefined, 'previous event record must exist'); - invariant(current !== undefined, 'current event record must exist'); - invariant( - current.seq === previous.seq + 1, - 'event log seq values must increase by 1 without gaps', - ); +function assertContiguousSequence(records: readonly EventRecord[]): void { + let previousSeq: number | undefined; + for (const record of records) { + assertSeqFollows(previousSeq, record.seq); + previousSeq = record.seq; } } @@ -111,3 +141,147 @@ export async function readEventLogRecords( const content = await readFile(filePath, 'utf8'); return parseEventLogContent(content); } + +type MarkerEventRecord = Extract; + +/** True for the marker that the host writes when the log stops. */ +export function isEventLogStopMarker( + record: EventRecord, +): record is MarkerEventRecord { + return ( + record.type === 'marker' && + record.payload.label.startsWith(EVENT_LOG_LIMIT_LABEL_PREFIX) + ); +} + +function stopOfMarker(marker: MarkerEventRecord): EventLogStop { + const limit = /^ \(\d+ (bytes|events)\)/.exec( + marker.payload.label.slice(EVENT_LOG_LIMIT_LABEL_PREFIX.length), + )?.[1]; + // The prefix is reserved, so only a log that agent-tty did not write holds + // a stop marker with no limit. The host still honours it as a stop. + return { + reason: limit === 'bytes' || limit === 'events' ? limit : 'unknown', + seq: marker.seq, + }; +} + +/** + * Returns where the log stopped: the `seq` of its first stop marker and the + * limit that the marker names. Returns `undefined` when `records` hold no + * stop marker. `EventLog.open` uses the same check to start a reopened log + * stopped. + */ +export function findEventLogStop( + records: readonly EventRecord[], +): EventLogStop | undefined { + const marker = records.find(isEventLogStopMarker); + return marker === undefined ? undefined : stopOfMarker(marker); +} + +export interface EventLogScan { + /** Non-blank lines of the file, valid or not. */ + readonly entryCount: number; + /** Bytes of the file that the scan read. */ + readonly byteCount: number; + /** See `scanEventLog`. */ + readonly stop: EventLogStop | undefined; +} + +/** + * Reads the file once and returns what it holds. All three results describe + * the same bytes: the file as it was when the scan measured it, so a live + * host that appends meanwhile cannot make them disagree. Returns `undefined` + * when the file does not exist. + * + * `stop` is the first stop marker that a replay would reach: the scan parses + * lines with the codec of `parseEventLogContent`, and it stops looking at the + * first line that is not a valid record with the next `seq`. A line that the + * live host has not finished writing ends the search the same way. So a log + * that replay rejects can still be counted here, and a marker after the + * first bad line is not reported. + * + * Unlike `readEventLogRecords`, it does not enforce `MAX_EVENT_LOG_SIZE`: it + * streams the file, so it also reads a log that an older version wrote past + * the limit. + */ +export async function scanEventLog( + filePath: string, +): Promise { + invariant(filePath.length > 0, 'filePath must be a non-empty string'); + + let file: FileHandle; + try { + file = await open(filePath, 'r'); + } catch (error: unknown) { + if (hasErrorCode(error, 'ENOENT')) { + return undefined; + } + + throw error; + } + + try { + const { size } = await file.stat(); + let entryCount = 0; + let stop: EventLogStop | undefined; + let previousSeq: number | undefined; + let searching = true; + + const readLine = (rawLine: string): void => { + const line = rawLine.trim(); + if (line.length === 0) { + return; + } + + entryCount += 1; + if (!searching) { + return; + } + + try { + const record = parseEventLogLine(line, entryCount); + assertSeqFollows(previousSeq, record.seq); + previousSeq = record.seq; + if (isEventLogStopMarker(record)) { + stop = stopOfMarker(record); + searching = false; + } + } catch { + searching = false; + } + }; + + if (size > 0) { + // Split on '\n' only, as parseEventLogContent does. readline also + // splits on U+2028 and U+2029, which JSON.stringify writes raw. The + // pieces of a long line are joined once, when its newline arrives. + let pending: string[] = []; + for await (const chunk of file.createReadStream({ + encoding: 'utf8', + start: 0, + end: size - 1, + autoClose: false, + })) { + const text = chunk as string; + let start = 0; + let newline = text.indexOf('\n', start); + while (newline !== -1) { + pending.push(text.slice(start, newline)); + readLine(pending.join('')); + pending = []; + start = newline + 1; + newline = text.indexOf('\n', start); + } + if (start < text.length) { + pending.push(text.slice(start)); + } + } + readLine(pending.join('')); + } + + return { entryCount, byteCount: size, stop }; + } finally { + await file.close(); + } +} diff --git a/test/fixtures/plant-stopped-event-log.ts b/test/fixtures/plant-stopped-event-log.ts new file mode 100644 index 00000000..f9f5f2b1 --- /dev/null +++ b/test/fixtures/plant-stopped-event-log.ts @@ -0,0 +1,28 @@ +// Preload (`--import`) for a `create` process. The host starts on an event log +// that holds one output record and the record that stops the log, as a host +// leaves behind a log that reached its limit. The host reopens that log +// stopped, so it records nothing that the program prints. Only the host runs +// this. `PLANT_STOP_LABEL` replaces the label of the stop record. +import { writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import process from 'node:process'; + +const hostIndex = process.argv.indexOf('_host'); +if (hostIndex !== -1) { + const home = process.env.AGENT_TTY_HOME; + const sessionId = process.argv[hostIndex + 1]; + if (home === undefined || sessionId === undefined) { + throw new Error('the host needs AGENT_TTY_HOME and a session id'); + } + const record = (seq: number, type: string, payload: object): string => + `${JSON.stringify({ seq, ts: new Date().toISOString(), type, payload })}\n`; + writeFileSync( + join(home, 'sessions', sessionId, 'events.jsonl'), + record(0, 'output', { data: 'planted output\r\n' }) + + record(1, 'marker', { + label: + process.env.PLANT_STOP_LABEL ?? + 'agent-tty: event log limit reached (250000 events); later events are not recorded', + }), + ); +} diff --git a/test/integration/event-log.test.ts b/test/integration/event-log.test.ts index dbe5921a..3fae4059 100644 --- a/test/integration/event-log.test.ts +++ b/test/integration/event-log.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, realpath } from 'node:fs/promises'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { join, resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; @@ -124,4 +125,277 @@ describe('event-log integration', { timeout: 30000 }, () => { destroySession(testHome, sessionId); } }); + + describe('a log that stopped', () => { + const NOTE = + 'The event log stopped at seq 1 (event limit), so output after it is not seen.'; + const env = (): Record => ({ AGENT_TTY_HOME: testHome }); + + // The host starts on a log that holds one output record and the stop + // record, so it records nothing that the program prints after that. + function createStoppedSession( + script: string, + plantedLabel?: string, + ): string { + const preload = pathToFileURL( + resolve('test/fixtures/plant-stopped-event-log.ts'), + ); + return createSession(testHome, ['/bin/sh', '-c', script], { + env: { + NODE_OPTIONS: [ + process.env.NODE_OPTIONS ?? '', + `--import=${preload.href}`, + ].join(' '), + ...(plantedLabel === undefined + ? {} + : { PLANT_STOP_LABEL: plantedLabel }), + }, + }); + } + + it('shows the stop in inspect, in a timed-out wait and in batch waits of a live session', async () => { + const sessionId = createStoppedSession( + "sleep 1; printf 'after the stop\\n'; exec cat", + ); + + try { + await sleep(1500); + + const inspectResult = runCli(['inspect', sessionId, '--json'], env()); + expect(inspectResult.status).toBe(0); + expect( + (JSON.parse(inspectResult.stdout) as { result: object }).result, + ).toMatchObject({ + eventCount: 2, + lastEventSeq: 1, + eventLogStop: { reason: 'events', seq: 1 }, + }); + + const humanResult = runCli(['inspect', sessionId], env()); + expect(humanResult.status).toBe(0); + expect(humanResult.stdout).toContain( + 'Event Log Stopped: at seq 1 (event limit); later events are not recorded', + ); + + // A wait that the stored log satisfies is not touched. + const matched = runCli( + ['wait', sessionId, '--text', 'planted output', '--json'], + env(), + 30_000, + ); + expect(matched.status).toBe(0); + expect(JSON.parse(matched.stdout) as object).toMatchObject({ + ok: true, + result: { matched: true, timedOut: false }, + }); + + // The host reports the stop with its timeout result. + const timedOut = runCli( + ['wait', sessionId, '--text', 'after the stop', '--timeout', '2000'], + env(), + 30_000, + ); + expect(timedOut.status).toBe(11); + expect(timedOut.stderr).toContain(NOTE); + + const timedOutJson = runCli( + [ + 'wait', + sessionId, + '--text', + 'after the stop', + '--timeout', + '2000', + '--json', + ], + env(), + 30_000, + ); + expect(timedOutJson.status).toBe(11); + const envelope = JSON.parse(timedOutJson.stdout) as { + error: { + code: string; + message: string; + details?: Record; + }; + result: { timedOut: boolean; eventLogStop?: unknown }; + }; + expect(envelope.error.code).toBe('WAIT_TIMEOUT'); + expect(envelope.error.message).toBe( + `Wait timed out before its condition was met. ${NOTE}`, + ); + expect(envelope.error.details?.eventLogStop).toEqual({ + reason: 'events', + seq: 1, + }); + expect(envelope.result).toMatchObject({ + timedOut: true, + eventLogStop: { reason: 'events', seq: 1 }, + }); + + const batchResult = runCli( + [ + 'batch', + sessionId, + JSON.stringify([ + { wait: { text: 'after the stop', timeout: 2000 } }, + ]), + '--json', + ], + env(), + 30_000, + ); + expect(batchResult.status).toBe(11); + expect( + (JSON.parse(batchResult.stdout) as { error: unknown }).error, + ).toMatchObject({ + code: 'WAIT_TIMEOUT', + message: `Render wait at step 0 timed out after 2000 ms. ${NOTE}`, + }); + + // Two failed waits, and no step reads the log again: each carries + // the sentence. + const keepGoing = runCli( + [ + 'batch', + sessionId, + JSON.stringify([ + { wait: { text: 'after the stop', timeout: 1500 } }, + { wait: { text: 'also after the stop', timeout: 1500 } }, + ]), + '--keep-going', + '--json', + ], + env(), + 60_000, + ); + expect(keepGoing.status).toBe(1); + const keepGoingResult = ( + JSON.parse(keepGoing.stdout) as { + result: { failedIndices: number[]; steps: { error?: object }[] }; + } + ).result; + expect(keepGoingResult.failedIndices).toEqual([0, 1]); + expect(keepGoingResult.steps.map((step) => step.error)).toEqual([ + expect.objectContaining({ + code: 'WAIT_TIMEOUT', + message: `Render wait at step 0 timed out after 1500 ms. ${NOTE}`, + }), + expect.objectContaining({ + code: 'WAIT_TIMEOUT', + message: `Render wait at step 1 timed out after 1500 ms. ${NOTE}`, + }), + ]); + } finally { + destroySession(testHome, sessionId); + } + }, 120_000); + + it('shows the stop for an exited session, also in a wait that has no host to ask', async () => { + const sessionId = createStoppedSession( + "sleep 1; printf 'after the stop\\n'", + ); + + try { + for (let attempt = 0; attempt < 40; attempt += 1) { + const status = ( + JSON.parse( + runCli(['inspect', sessionId, '--json'], env()).stdout, + ) as { result: { session: { status: string } } } + ).result.session.status; + if (status === 'exited') { + break; + } + await sleep(250); + } + + const inspectResult = runCli(['inspect', sessionId, '--json'], env()); + expect(inspectResult.status).toBe(0); + expect( + (JSON.parse(inspectResult.stdout) as { result: object }).result, + ).toMatchObject({ + session: { status: 'exited' }, + eventLogStop: { reason: 'events', seq: 1 }, + }); + + const matched = runCli( + ['wait', sessionId, '--text', 'planted output', '--json'], + env(), + 30_000, + ); + expect(matched.status).toBe(0); + + const later = runCli( + ['wait', sessionId, '--text', 'after the stop', '--json'], + env(), + 30_000, + ); + expect(later.status).toBe(10); + const { error } = JSON.parse(later.stdout) as { + error: { + code: string; + message: string; + details?: Record; + }; + }; + expect(error.code).toBe('REPLAY_ERROR'); + expect(error.message).toBe( + `Session host became unreachable during render wait, and the latest offline snapshot did not satisfy the requested wait condition. ${NOTE}`, + ); + expect(error.details?.eventLogStop).toEqual({ + reason: 'events', + seq: 1, + }); + } finally { + destroySession(testHome, sessionId); + } + }, 120_000); + + it('shows a stop marker that names no limit with the reason unknown instead of failing', async () => { + const sessionId = createStoppedSession( + "sleep 1; printf 'after the stop\\n'; exec cat", + 'agent-tty: event log limit reached (soon)', + ); + + try { + await sleep(1500); + + const inspectResult = runCli(['inspect', sessionId, '--json'], env()); + expect(inspectResult.status).toBe(0); + expect( + (JSON.parse(inspectResult.stdout) as { result: object }).result, + ).toMatchObject({ eventLogStop: { reason: 'unknown', seq: 1 } }); + + const humanResult = runCli(['inspect', sessionId], env()); + expect(humanResult.status).toBe(0); + expect(humanResult.stdout).toContain( + 'Event Log Stopped: at seq 1 (reason unknown); later events are not recorded', + ); + + const timedOut = runCli( + [ + 'wait', + sessionId, + '--text', + 'after the stop', + '--timeout', + '2000', + '--json', + ], + env(), + 30_000, + ); + expect(timedOut.status).toBe(11); + expect( + (JSON.parse(timedOut.stdout) as { error: unknown }).error, + ).toMatchObject({ + code: 'WAIT_TIMEOUT', + message: + 'Wait timed out before its condition was met. The event log stopped at seq 1 (reason unknown), so output after it is not seen.', + }); + } finally { + destroySession(testHome, sessionId); + } + }, 120_000); + }); }); diff --git a/test/unit/batch/executor.test.ts b/test/unit/batch/executor.test.ts index 49ad54cb..912651c2 100644 --- a/test/unit/batch/executor.test.ts +++ b/test/unit/batch/executor.test.ts @@ -480,6 +480,82 @@ describe('executeBatch', () => { }); }); + it('names where the host reports that its event log stopped in the message of a timed-out wait', async () => { + const { driver } = createFakeDriver({ + waitResults: [ + { + matched: false, + timedOut: true, + capturedAtSeq: 5, + eventLogStop: { reason: 'events', seq: 5 }, + }, + ], + }); + const result = await executeBatch({ + plan: plan([{ wait: { text: 'never', timeout: 1000 } }]), + driver, + keepGoing: false, + }); + + expect(result.steps[0]).toMatchObject({ + status: 'failed', + timedOut: true, + error: { + code: 'WAIT_TIMEOUT', + message: + 'Render wait at step 0 timed out after 1000 ms. The event log stopped at seq 5 (event limit), so output after it is not seen.', + }, + }); + }); + + it('names a stop in each timed-out wait of a keep-going batch', async () => { + const stopped = { + matched: false, + timedOut: true, + capturedAtSeq: 5, + eventLogStop: { reason: 'bytes' as const, seq: 5 }, + }; + const { driver } = createFakeDriver({ + waitResults: [stopped, stopped], + }); + const result = await executeBatch({ + plan: plan([ + { wait: { text: 'one', timeout: 1000 } }, + { wait: { text: 'two', timeout: 1000 } }, + ]), + driver, + keepGoing: true, + }); + + expect(result.failedIndices).toEqual([0, 1]); + expect( + result.steps.map((step) => + step.status === 'failed' ? step.error?.message : undefined, + ), + ).toEqual([ + 'Render wait at step 0 timed out after 1000 ms. The event log stopped at seq 5 (byte limit), so output after it is not seen.', + 'Render wait at step 1 timed out after 1000 ms. The event log stopped at seq 5 (byte limit), so output after it is not seen.', + ]); + }); + + it('keeps the message of a timed-out wait when the host reports no stop', async () => { + const { driver } = createFakeDriver({ + waitResults: [{ matched: false, timedOut: true, capturedAtSeq: 5 }], + }); + const result = await executeBatch({ + plan: plan([{ wait: { text: 'never', timeout: 1000 } }]), + driver, + keepGoing: false, + }); + + expect(result.steps[0]).toMatchObject({ + error: { + code: 'WAIT_TIMEOUT', + message: 'Render wait at step 0 timed out after 1000 ms.', + }, + }); + }); + it('treats an unmatched-but-not-timedOut wait as a WAIT_TIMEOUT failure', async () => { const { driver } = createFakeDriver({ waitResults: [{ matched: false, timedOut: false, capturedAtSeq: 1 }], diff --git a/test/unit/commands/batch.test.ts b/test/unit/commands/batch.test.ts index 83bf7907..37dc4ac5 100644 --- a/test/unit/commands/batch.test.ts +++ b/test/unit/commands/batch.test.ts @@ -1,8 +1,35 @@ -import { describe, expect, it } from 'vitest'; +import process from 'node:process'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { BatchResult } from '../../../src/batch/result.js'; -import { buildBatchLines } from '../../../src/cli/commands/batch.js'; +import { + buildBatchLines, + runBatchCommand, +} from '../../../src/cli/commands/batch.js'; + +const mocks = vi.hoisted(() => ({ + resolveCommandTarget: vi.fn(), + sendRpc: vi.fn(), + emitSuccess: vi.fn(), + emitFailure: vi.fn(), + emitSuccessSync: vi.fn(), +})); + +vi.mock('../../../src/cli/commandTarget.js', () => ({ + resolveCommandTarget: mocks.resolveCommandTarget, +})); + +vi.mock('../../../src/host/rpcClient.js', () => ({ + sendRpc: mocks.sendRpc, +})); + +vi.mock('../../../src/cli/output.js', () => ({ + emitSuccess: mocks.emitSuccess, + emitFailure: mocks.emitFailure, + emitSuccessSync: mocks.emitSuccessSync, +})); describe('buildBatchLines', () => { it('labels an interrupted step as interrupted, not as a success', () => { @@ -33,3 +60,100 @@ describe('buildBatchLines', () => { expect(lines.at(-1)).toBe('1/3 steps completed'); }); }); + +describe('runBatchCommand when the host reports that its event log stopped', () => { + const stopNote = + 'The event log stopped at seq 5 (byte limit), so output after it is not seen.'; + const timedOutWait = { + matched: false, + timedOut: true, + capturedAtSeq: 5, + eventLogStop: { reason: 'bytes', seq: 5 }, + }; + const options = { + context: {} as Parameters[0]['context'], + json: true, + sessionId: 'session-01', + steps: JSON.stringify([ + { wait: { text: 'one', timeout: 1000 } }, + { wait: { text: 'two', timeout: 1000 } }, + ]), + keepGoing: true, + }; + + beforeEach(() => { + vi.clearAllMocks(); + mocks.resolveCommandTarget.mockResolvedValue({ + sessionId: 'session-01', + sessionDirectory: '/tmp/agent-tty/sessions/session-01', + socketPath: '/tmp/agent-tty/sessions/session-01/rpc.sock', + }); + }); + + afterEach(() => { + process.exitCode = undefined; + }); + + it('names the stop in every timed-out wait of a --keep-going batch', async () => { + mocks.sendRpc.mockResolvedValue(timedOutWait); + + await runBatchCommand(options); + + expect(mocks.sendRpc).toHaveBeenCalledTimes(2); + const methods = (mocks.sendRpc.mock.calls as unknown[][]).map( + (call) => call[1], + ); + expect(methods).toEqual(['waitForRender', 'waitForRender']); + const emitted = mocks.emitSuccess.mock.calls[0] as [ + { result: BatchResult }, + ]; + const { result } = emitted[0]; + expect(result.failedIndices).toEqual([0, 1]); + expect( + result.steps.map((step) => + step.status === 'failed' ? step.error : undefined, + ), + ).toEqual([ + { + code: 'WAIT_TIMEOUT', + message: `Render wait at step 0 timed out after 1000 ms. ${stopNote}`, + }, + { + code: 'WAIT_TIMEOUT', + message: `Render wait at step 1 timed out after 1000 ms. ${stopNote}`, + }, + ]); + }); + + it('names the stop in the error of a failed batch under --json', async () => { + mocks.sendRpc.mockResolvedValue(timedOutWait); + + await runBatchCommand({ ...options, keepGoing: false }); + + expect(mocks.sendRpc).toHaveBeenCalledTimes(1); + const emitted = mocks.emitFailure.mock.calls[0] as [ + { error: { code: string; message: string } }, + ]; + expect(emitted[0].error).toMatchObject({ + code: 'WAIT_TIMEOUT', + message: `Render wait at step 0 timed out after 1000 ms. ${stopNote}`, + }); + }); + + it('keeps the message of a timed-out wait when the host reports no stop', async () => { + mocks.sendRpc.mockResolvedValue({ + matched: false, + timedOut: true, + capturedAtSeq: 5, + }); + + await runBatchCommand({ ...options, keepGoing: false }); + + const emitted = mocks.emitFailure.mock.calls[0] as [ + { error: { message: string } }, + ]; + expect(emitted[0].error.message).toBe( + 'Render wait at step 0 timed out after 1000 ms.', + ); + }); +}); diff --git a/test/unit/commands/golden-envelopes.test.ts b/test/unit/commands/golden-envelopes.test.ts index 49beb458..6b4b34d5 100644 --- a/test/unit/commands/golden-envelopes.test.ts +++ b/test/unit/commands/golden-envelopes.test.ts @@ -1305,6 +1305,27 @@ describe('JSON envelope contracts', () => { expect(InspectResultSchema.safeParse(result).success).toBe(true); }); + it('locks the inspect success envelope shape when the event log stopped', () => { + const result = InspectResultSchema.parse({ + session: createSessionRecord(), + eventCount: 250_000, + uptime: 1_000, + lastEventSeq: 249_999, + terminationCategory: 'running', + usedOfflineReplay: false, + rendererRuntime: { + backend: 'ghostty-web', + mode: 'live-host', + status: 'healthy', + }, + eventLogBytes: 21_400_000, + eventLogStop: { reason: 'events', seq: 249_998 }, + }); + + expectLockedSuccessEnvelope('inspect', result); + expect(result.eventLogStop).toEqual({ reason: 'events', seq: 249_998 }); + }); + it('accepts inspect result with offline renderer runtime', () => { const result = { session: createSessionRecord(), @@ -1571,4 +1592,50 @@ describe('JSON envelope contracts', () => { result: { matched: false, timedOut: true, capturedAtSeq: 7 }, }); }); + + it('locks the WAIT_TIMEOUT error envelope shape when the event log stopped', () => { + const eventLogStop = { reason: 'bytes', seq: 7 } as const; + const result = WaitForRenderResultSchema.parse({ + matched: false, + timedOut: true, + capturedAtSeq: 7, + eventLogStop, + }); + const error = makeCliError(ERROR_CODES.WAIT_TIMEOUT, { + message: + 'Wait timed out before its condition was met. The event log stopped at seq 7 (byte limit), so output after it is not seen.', + details: { sessionId: 'session-01', timeoutMs: 1000, eventLogStop }, + }); + + const envelope: CommandEnvelope = + createErrorEnvelope( + 'wait', + { + code: error.code, + message: error.message, + retryable: error.retryable, + details: error.details, + }, + result, + ); + + expect(envelope).toEqual({ + ok: false, + command: 'wait', + timestamp: LOCKED_TIMESTAMP, + error: { + code: 'WAIT_TIMEOUT', + message: + 'Wait timed out before its condition was met. The event log stopped at seq 7 (byte limit), so output after it is not seen.', + retryable: false, + details: { sessionId: 'session-01', timeoutMs: 1000, eventLogStop }, + }, + result: { + matched: false, + timedOut: true, + capturedAtSeq: 7, + eventLogStop, + }, + }); + }); }); diff --git a/test/unit/commands/inspect.test.ts b/test/unit/commands/inspect.test.ts index afc02b09..d4df4891 100644 --- a/test/unit/commands/inspect.test.ts +++ b/test/unit/commands/inspect.test.ts @@ -4,8 +4,7 @@ import { ERROR_CODES, makeCliError } from '../../../src/protocol/errors.js'; const mocks = vi.hoisted(() => ({ computeArtifactHealth: vi.fn(), - countEventLogEntries: vi.fn(), - statEventLogBytes: vi.fn(), + scanEventLog: vi.fn(), deriveTerminationCategory: vi.fn(), emitSuccess: vi.fn(), reconcileSession: vi.fn(), @@ -19,13 +18,24 @@ const mocks = vi.hoisted(() => ({ socketPath: vi.fn(), })); +// What `scanEventLog` returns for a log of `entryCount` records that has not +// stopped. +function scanOf( + entryCount: number, + overrides: { + byteCount?: number; + stop?: { reason: string; seq: number }; + } = {}, +) { + return { entryCount, byteCount: 1024, stop: undefined, ...overrides }; +} + vi.mock('../../../src/cli/output.js', () => ({ emitSuccess: mocks.emitSuccess, })); -vi.mock('../../../src/host/eventLog.js', () => ({ - countEventLogEntries: mocks.countEventLogEntries, - statEventLogBytes: mocks.statEventLogBytes, +vi.mock('../../../src/storage/eventLogCodec.js', () => ({ + scanEventLog: mocks.scanEventLog, })); vi.mock('../../../src/host/lifecycle.js', () => ({ @@ -133,8 +143,7 @@ describe('inspect command', () => { (sessionDirectory: string) => `${sessionDirectory}/rpc.sock`, ); mocks.computeArtifactHealth.mockResolvedValue(DEFAULT_ARTIFACT_HEALTH); - mocks.countEventLogEntries.mockResolvedValue(2); - mocks.statEventLogBytes.mockResolvedValue(undefined); + mocks.scanEventLog.mockResolvedValue(scanOf(2)); mocks.deriveTerminationCategory.mockReturnValue('running'); mocks.readManifestIfExists.mockResolvedValue( createSessionRecord('running'), @@ -161,7 +170,7 @@ describe('inspect command', () => { '/tmp/agent-tty/sessions/session-01/rpc.sock', 'inspect', ); - expect(mocks.countEventLogEntries).toHaveBeenCalledWith( + expect(mocks.scanEventLog).toHaveBeenCalledWith( '/tmp/agent-tty/sessions/session-01/events.jsonl', ); expect(mocks.computeArtifactHealth).toHaveBeenCalledWith( @@ -220,6 +229,123 @@ describe('inspect command', () => { expect(emitted.lines).not.toContain('Termination: running'); }); + it('leaves eventLogStop and its line out when the stored log has not stopped', async () => { + mocks.sendRpc.mockResolvedValue({ session: createSessionRecord() }); + + await runInspectCommand({ + context: TEST_CONTEXT, + json: true, + sessionId: 'session-01', + }); + + const emitted = getLastEmitSuccessPayload() as { + result: object; + lines: string[]; + }; + expect(emitted.result).not.toHaveProperty('eventLogStop'); + expect(emitted.lines.join('\n')).not.toContain('Event Log Stopped'); + }); + + it.each([ + ['bytes', 'byte limit'], + ['events', 'event limit'], + ['unknown', 'reason unknown'], + ] as const)( + 'reports the stop of the stored log with the %s reason, from a live host', + async (reason, words) => { + mocks.sendRpc.mockResolvedValue({ session: createSessionRecord() }); + mocks.scanEventLog.mockResolvedValue( + scanOf(42, { stop: { reason, seq: 41 } }), + ); + + await runInspectCommand({ + context: TEST_CONTEXT, + json: true, + sessionId: 'session-01', + }); + + const emitted = getLastEmitSuccessPayload() as { + result: { eventLogStop?: unknown }; + lines: string[]; + }; + expect(emitted.result.eventLogStop).toEqual({ reason, seq: 41 }); + expect(emitted.lines).toContain( + `Event Log Stopped: at seq 41 (${words}); later events are not recorded`, + ); + }, + ); + + it('reports the stop of the stored log when the host is unreachable', async () => { + mocks.sendRpc.mockRejectedValue( + makeCliError(ERROR_CODES.HOST_UNREACHABLE, { + message: 'Session host is unreachable.', + }), + ); + mocks.readManifest.mockResolvedValue(createSessionRecord('exited')); + mocks.scanEventLog.mockResolvedValue( + scanOf(8, { stop: { reason: 'bytes', seq: 7 } }), + ); + + await runInspectCommand({ + context: TEST_CONTEXT, + json: true, + sessionId: 'session-01', + }); + + const emitted = getLastEmitSuccessPayload() as { + result: { usedOfflineReplay?: boolean; eventLogStop?: unknown }; + }; + expect(emitted.result.usedOfflineReplay).toBe(true); + expect(emitted.result.eventLogStop).toEqual({ reason: 'bytes', seq: 7 }); + }); + + it('takes the count, the size and the stop from one read of the log', async () => { + mocks.sendRpc.mockResolvedValue({ session: createSessionRecord() }); + mocks.scanEventLog.mockResolvedValue( + scanOf(8, { byteCount: 5000, stop: { reason: 'events', seq: 7 } }), + ); + + await runInspectCommand({ + context: TEST_CONTEXT, + json: true, + sessionId: 'session-01', + }); + + expect(mocks.scanEventLog).toHaveBeenCalledTimes(1); + const { result } = getLastEmitSuccessPayload() as { + result: { + eventCount: number; + lastEventSeq: number; + eventLogBytes: number; + eventLogStop: { seq: number }; + }; + }; + expect(result).toMatchObject({ + eventCount: 8, + lastEventSeq: 7, + eventLogBytes: 5000, + eventLogStop: { reason: 'events', seq: 7 }, + }); + // The stop marker is one of the counted records. + expect(result.eventLogStop.seq).toBeLessThanOrEqual(result.lastEventSeq); + }); + + it('shows an empty log and no size when the log file is missing', async () => { + mocks.sendRpc.mockResolvedValue({ session: createSessionRecord() }); + mocks.scanEventLog.mockResolvedValue(undefined); + + await runInspectCommand({ + context: TEST_CONTEXT, + json: true, + sessionId: 'session-01', + }); + + const { result } = getLastEmitSuccessPayload() as { result: object }; + expect(result).toMatchObject({ eventCount: 0, lastEventSeq: undefined }); + expect(result).not.toHaveProperty('eventLogBytes'); + expect(result).not.toHaveProperty('eventLogStop'); + }); + it('treats exiting sessions as live-host renderer runtime', async () => { const exitingSession = createSessionRecord('exiting'); mocks.readManifestIfExists.mockResolvedValueOnce(exitingSession); @@ -268,7 +394,7 @@ describe('inspect command', () => { new Error('EACCES: permission denied'), ); mocks.sendRpc.mockResolvedValueOnce({ session: liveSession }); - mocks.countEventLogEntries.mockResolvedValueOnce(5); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(5)); mocks.deriveTerminationCategory.mockReturnValueOnce('running'); await runInspectCommand({ @@ -353,7 +479,7 @@ describe('inspect command', () => { expect(mocks.readManifest).toHaveBeenCalledWith( '/tmp/agent-tty/sessions/session-01/session.json', ); - expect(mocks.countEventLogEntries).toHaveBeenCalledWith( + expect(mocks.scanEventLog).toHaveBeenCalledWith( '/tmp/agent-tty/sessions/session-01/events.jsonl', ); expect(mocks.computeArtifactHealth).toHaveBeenCalledWith( @@ -420,7 +546,7 @@ describe('inspect command', () => { ...failedSession, status: 'destroyed', }); - mocks.countEventLogEntries.mockResolvedValueOnce(3); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(3)); await runInspectCommand({ context: TEST_CONTEXT, @@ -446,7 +572,7 @@ describe('inspect command', () => { message: 'disk full', }), ); - mocks.countEventLogEntries.mockResolvedValueOnce(3); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(3)); await runInspectCommand({ context: TEST_CONTEXT, @@ -471,7 +597,7 @@ describe('inspect command', () => { code: 'EPERM', }), ); - mocks.countEventLogEntries.mockResolvedValueOnce(3); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(3)); await runInspectCommand({ context: TEST_CONTEXT, @@ -488,7 +614,7 @@ describe('inspect command', () => { it('reports offline replay renderer runtime for exited sessions', async () => { const exitedSession = createSessionRecord('exited'); mocks.readManifestIfExists.mockResolvedValueOnce(exitedSession); - mocks.countEventLogEntries.mockResolvedValueOnce(3); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(3)); mocks.deriveTerminationCategory.mockReturnValueOnce('clean-exit'); await runInspectCommand({ @@ -528,7 +654,7 @@ describe('inspect command', () => { it('reports offline replay renderer runtime for destroying sessions', async () => { const destroyingSession = createSessionRecord('destroying'); mocks.readManifestIfExists.mockResolvedValueOnce(destroyingSession); - mocks.countEventLogEntries.mockResolvedValueOnce(4); + mocks.scanEventLog.mockResolvedValueOnce(scanOf(4)); mocks.deriveTerminationCategory.mockReturnValueOnce('destroyed'); await runInspectCommand({ @@ -567,7 +693,7 @@ describe('inspect command', () => { it('omits last event sequence from human output when the event log is empty', async () => { const liveSession = createSessionRecord('running'); - mocks.countEventLogEntries.mockResolvedValue(0); + mocks.scanEventLog.mockResolvedValue(scanOf(0)); mocks.sendRpc.mockResolvedValue({ session: liveSession }); await runInspectCommand({ @@ -720,7 +846,7 @@ describe('inspect command', () => { it('surfaces eventLogBytes in both live and offline modes', async () => { const liveSession = createSessionRecord('running'); mocks.sendRpc.mockResolvedValue({ session: liveSession }); - mocks.statEventLogBytes.mockResolvedValue(4096); + mocks.scanEventLog.mockResolvedValue(scanOf(2, { byteCount: 4096 })); await runInspectCommand({ context: TEST_CONTEXT, @@ -742,7 +868,7 @@ describe('inspect command', () => { }), ); mocks.readManifest.mockResolvedValue(reconciledSession); - mocks.statEventLogBytes.mockResolvedValue(8192); + mocks.scanEventLog.mockResolvedValue(scanOf(2, { byteCount: 8192 })); await runInspectCommand({ context: TEST_CONTEXT, diff --git a/test/unit/commands/wait.test.ts b/test/unit/commands/wait.test.ts index 299807a3..6f9b3409 100644 --- a/test/unit/commands/wait.test.ts +++ b/test/unit/commands/wait.test.ts @@ -103,13 +103,14 @@ function createOptions( function mockOfflineReplaySnapshot( snapshotOverrides: Parameters[0] = {}, + events: unknown[] = [], ): void { mocks.withOfflineReplayRenderer.mockImplementation( async ( _options: unknown, run: (context: { manifest: ReturnType; - replayInput: Record; + replayInput: { events: unknown[] }; backend: { snapshot: (options?: unknown) => Promise; }; @@ -123,7 +124,7 @@ function mockOfflineReplaySnapshot( return run({ manifest: createSessionRecord('exited', 0), - replayInput: {}, + replayInput: { events }, backend: mockBackend, }); }, @@ -640,6 +641,248 @@ describe('wait command', () => { }); }); + describe('when the event log stopped', () => { + const renderTimeout = { matched: false, timedOut: true, capturedAtSeq: 1 }; + const stopped = { reason: 'bytes', seq: 1 }; + const stopNote = + 'The event log stopped at seq 1 (byte limit), so output after it is not seen.'; + const plainTimeout = 'Wait timed out before its condition was met.'; + + const stopMarker = (seq: number, label: string) => ({ + seq, + ts: '2026-03-19T12:00:00.000Z', + type: 'marker', + payload: { label }, + }); + const stopEvents = [ + { + seq: 0, + ts: '2026-03-19T12:00:00.000Z', + type: 'output', + payload: { data: 'hello' }, + }, + stopMarker( + 1, + 'agent-tty: event log limit reached (52428800 bytes); later events are not recorded', + ), + ]; + + it.each([ + ['--text', { text: 'later' }, 30_000], + ['--regex', { regex: 'later.*' }, 30_000], + ['--screen-stable-ms', { screenStableMs: 1000, timeout: 5000 }, 5000], + ['--cursor-row', { cursorRow: 3 }, 30_000], + ['--after-seq', { text: 'later', afterSeq: 0 }, 30_000], + ])( + 'names the stop that the host reports in the WAIT_TIMEOUT of a %s wait', + async (_flag, flags, timeoutMs) => { + const result = { ...renderTimeout, eventLogStop: stopped }; + mocks.sendRpc.mockResolvedValue(result); + + await expect( + runWaitCommand(createOptions(flags)), + ).rejects.toMatchObject({ + code: ERROR_CODES.WAIT_TIMEOUT, + message: `${plainTimeout} ${stopNote}`, + details: { + sessionId: 'session-01', + timeoutMs, + eventLogStop: stopped, + }, + result, + }); + }, + ); + + it('names the stop under --json too', async () => { + mocks.sendRpc.mockResolvedValue({ + ...renderTimeout, + eventLogStop: stopped, + }); + + await expect( + runWaitCommand(createOptions({ json: true, text: 'later' })), + ).rejects.toMatchObject({ + code: ERROR_CODES.WAIT_TIMEOUT, + message: `${plainTimeout} ${stopNote}`, + }); + }); + + it('adds the note after the timeout that a human message names', async () => { + vi.useFakeTimers({ toFake: ['Date', 'performance'] }); + try { + mocks.sendRpc.mockImplementation(() => { + vi.advanceTimersByTime(1000); + return Promise.resolve({ ...renderTimeout, eventLogStop: stopped }); + }); + + await expect( + runWaitCommand(createOptions({ text: 'later', timeout: 1000 })), + ).rejects.toMatchObject({ + code: ERROR_CODES.WAIT_TIMEOUT, + message: `Wait timed out after 1000 ms. ${stopNote}`, + }); + } finally { + vi.useRealTimers(); + } + }); + + it.each([ + ['bytes', 'byte limit'], + ['events', 'event limit'], + ['unknown', 'reason unknown'], + ])('words the %s reason as "%s"', async (reason, words) => { + mocks.sendRpc.mockResolvedValue({ + ...renderTimeout, + eventLogStop: { reason, seq: 7 }, + }); + + await expect( + runWaitCommand(createOptions({ text: 'later' })), + ).rejects.toMatchObject({ + message: `${plainTimeout} The event log stopped at seq 7 (${words}), so output after it is not seen.`, + details: { eventLogStop: { reason, seq: 7 } }, + }); + }); + + it('keeps the plain WAIT_TIMEOUT when the host reports no stop', async () => { + mocks.sendRpc.mockResolvedValue(renderTimeout); + + const error = (await runWaitCommand( + createOptions({ text: 'later' }), + ).catch((caught: unknown) => caught)) as { + message: string; + details: object; + }; + expect(error.message).toBe(plainTimeout); + expect(error.details).toEqual({ + sessionId: 'session-01', + timeoutMs: 30_000, + }); + }); + + it('leaves a wait that matched as it is', async () => { + const result = { + matched: true, + timedOut: false, + matchedText: 'hello', + capturedAtSeq: 1, + }; + mocks.sendRpc.mockResolvedValue(result); + + await runWaitCommand(createOptions({ text: 'hello' })); + + expect(mocks.emitSuccess).toHaveBeenCalledWith( + expect.objectContaining({ command: 'wait', result }), + ); + }); + + it.each([ + ['--idle-ms', { idleMs: 500, timeout: 1000 }], + ['--exit', { waitForExit: true, timeout: 1000 }], + ])( + 'leaves the WAIT_TIMEOUT of a %s wait as it is', + async (_flag, flags) => { + mocks.sendRpc.mockResolvedValue({ timedOut: true }); + + const error = (await runWaitCommand(createOptions(flags)).catch( + (caught: unknown) => caught, + )) as { code: string; message: string; details: object }; + expect(error.code).toBe(ERROR_CODES.WAIT_TIMEOUT); + expect(error.message).not.toContain('event log'); + expect(error.details).toEqual({ + sessionId: 'session-01', + timeoutMs: 1000, + }); + }, + ); + + describe('offline, after the host became unreachable', () => { + beforeEach(() => { + mocks.sendRpc.mockRejectedValue( + makeCliError(ERROR_CODES.HOST_UNREACHABLE, { + message: 'Session host is unreachable.', + }), + ); + }); + + it('names the stop of the replayed events when the snapshot does not satisfy the wait', async () => { + mockOfflineReplaySnapshot( + { + capturedAtSeq: 1, + visibleLines: [{ row: 0, text: 'hello' }], + }, + stopEvents, + ); + + await expect( + runWaitCommand(createOptions({ text: 'later' })), + ).rejects.toMatchObject({ + code: ERROR_CODES.REPLAY_ERROR, + message: `Session host became unreachable during render wait, and the latest offline snapshot did not satisfy the requested wait condition. ${stopNote}`, + details: { + text: 'later', + capturedAtSeq: 1, + visibleLines: ['hello'], + eventLogStop: stopped, + }, + }); + }); + + it('names a stop whose label names no limit', async () => { + mockOfflineReplaySnapshot({ capturedAtSeq: 0 }, [ + stopMarker(0, 'agent-tty: event log limit reached (soon)'), + ]); + + const error = (await runWaitCommand( + createOptions({ text: 'later' }), + ).catch((caught: unknown) => caught)) as { + message: string; + details: object; + }; + expect( + error.message.endsWith( + 'The event log stopped at seq 0 (reason unknown), so output after it is not seen.', + ), + ).toBe(true); + expect(error.details).toMatchObject({ + eventLogStop: { reason: 'unknown', seq: 0 }, + }); + }); + + it('keeps the plain REPLAY_ERROR when the replayed events hold no stop', async () => { + mockOfflineReplaySnapshot( + { capturedAtSeq: 0, visibleLines: [{ row: 0, text: 'hello' }] }, + [stopEvents[0]], + ); + + const error = (await runWaitCommand( + createOptions({ text: 'later' }), + ).catch((caught: unknown) => caught)) as { + message: string; + details: object; + }; + expect(error.message).not.toContain('event log'); + expect(error.details).not.toHaveProperty('eventLogStop'); + }); + + it('does not name the stop when the snapshot satisfies the wait', async () => { + mockOfflineReplaySnapshot( + { capturedAtSeq: 1, visibleLines: [{ row: 0, text: 'hello' }] }, + stopEvents, + ); + + await runWaitCommand(createOptions({ text: 'hello' })); + + const emitted = mocks.emitSuccess.mock.calls[0] as [ + { command: string; result: object }, + ]; + expect(emitted[0].command).toBe('wait'); + expect(emitted[0].result).not.toHaveProperty('eventLogStop'); + }); + }); + }); + it('keeps the WAIT_TIMEOUT message of the JSON envelope', async () => { mocks.sendRpc.mockResolvedValue({ timedOut: true }); diff --git a/test/unit/host/eventLog.test.ts b/test/unit/host/eventLog.test.ts index bd5ca288..eca2ee61 100644 --- a/test/unit/host/eventLog.test.ts +++ b/test/unit/host/eventLog.test.ts @@ -14,16 +14,14 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { generateAsciicast } from '../../../src/export/asciicast.js'; import { - countEventLogEntries, - EVENT_LOG_LIMIT_LABEL_PREFIX, EVENT_LOG_TAIL_RESERVE_BYTES, EVENT_LOG_TAIL_RESERVE_ENTRIES, EventLog, MAX_EVENT_BUFFER_ENTRIES, - statEventLogBytes, } from '../../../src/host/eventLog.js'; import type { SessionRecord } from '../../../src/protocol/schemas.js'; import { + EVENT_LOG_LIMIT_LABEL_PREFIX, MAX_EVENT_LOG_SIZE, readEventLogRecords, } from '../../../src/storage/eventLogCodec.js'; @@ -31,94 +29,6 @@ import { let tempDir = ''; let eventLogPath = ''; -describe('countEventLogEntries', () => { - beforeEach(async () => { - // oxfmt-ignore - tempDir = await realpath(await mkdtemp(join(tmpdir(), 'agent-tty-event-log-'))); - eventLogPath = join(tempDir, 'events.jsonl'); - }); - - afterEach(async () => { - if (tempDir.length > 0) { - await rm(tempDir, { recursive: true, force: true }); - } - }); - - it('returns 0 when the event log file is missing', async () => { - await expect(countEventLogEntries(eventLogPath)).resolves.toBe(0); - }); - - it('returns 0 for an empty event log file', async () => { - await writeFile(eventLogPath, '', 'utf8'); - - await expect(countEventLogEntries(eventLogPath)).resolves.toBe(0); - }); - - it('counts non-empty lines in a JSONL event log', async () => { - await writeFile( - eventLogPath, - [ - JSON.stringify({ seq: 0, type: 'output' }), - '', - ' ', - JSON.stringify({ seq: 1, type: 'resize' }), - ].join('\n'), - 'utf8', - ); - - await expect(countEventLogEntries(eventLogPath)).resolves.toBe(2); - }); -}); - -describe('statEventLogBytes', () => { - beforeEach(async () => { - // oxfmt-ignore - tempDir = await realpath(await mkdtemp(join(tmpdir(), 'agent-tty-event-log-'))); - eventLogPath = join(tempDir, 'events.jsonl'); - }); - - afterEach(async () => { - if (tempDir.length > 0) { - await rm(tempDir, { recursive: true, force: true }); - } - }); - - it('returns undefined when the event log file is missing', async () => { - await expect(statEventLogBytes(eventLogPath)).resolves.toBeUndefined(); - }); - - it('returns 0 for an empty event log file', async () => { - await writeFile(eventLogPath, '', 'utf8'); - - await expect(statEventLogBytes(eventLogPath)).resolves.toBe(0); - }); - - it('returns the byte size of a populated event log file', async () => { - const payload = `${JSON.stringify({ seq: 0, type: 'output' })}\n`; - await writeFile(eventLogPath, payload, 'utf8'); - - await expect(statEventLogBytes(eventLogPath)).resolves.toBe( - Buffer.byteLength(payload, 'utf8'), - ); - }); - - it('rejects empty file paths', async () => { - await expect(statEventLogBytes('')).rejects.toThrow(); - }); - - it('rethrows non-ENOENT errors so callers see them', async () => { - // A path whose parent is a regular file rather than a directory triggers - // ENOTDIR from stat(), which must propagate (only ENOENT may be swallowed). - const blocker = join(tempDir, 'blocker'); - await writeFile(blocker, 'x', 'utf8'); - const childOfFile = join(blocker, 'events.jsonl'); - - await expect(statEventLogBytes(childOfFile)).rejects.toMatchObject({ - code: 'ENOTDIR', - }); - }); -}); - describe('EventLog', () => { beforeEach(async () => { // oxfmt-ignore @@ -1339,3 +1249,133 @@ describe('EventLog size limit', () => { ).rejects.toThrow('maxEntries must leave room'); }); }); + +describe('EventLog.getStop', () => { + const LIMIT = EVENT_LOG_TAIL_RESERVE_BYTES * 2; + + beforeEach(async () => { + // oxfmt-ignore + tempDir = await realpath(await mkdtemp(join(tmpdir(), 'agent-tty-event-log-stop-'))); + eventLogPath = join(tempDir, 'events.jsonl'); + await writeFile(eventLogPath, '', 'utf8'); + }); + + afterEach(async () => { + if (tempDir.length > 0) { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it('returns undefined until the log stops', async () => { + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + expect(eventLog.getStop()).toBeUndefined(); + await eventLog.append('output', { data: 'hello' }); + await eventLog.append('marker', { label: 'a user marker' }); + expect(eventLog.getStop()).toBeUndefined(); + } finally { + await eventLog.close(); + } + }); + + it('names the byte limit and the seq of the stop marker', async () => { + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + await eventLog.append('output', { data: 'prompt$ ' }); + await eventLog.append('output', { data: 'x'.repeat(6000) }); + await eventLog.append('exit', { exitCode: 0, exitSignal: null }); + + expect(eventLog.getEvents().map((event) => event.type)).toEqual([ + 'output', + 'marker', + 'exit', + ]); + expect(eventLog.getStop()).toEqual({ reason: 'bytes', seq: 1 }); + } finally { + await eventLog.close(); + } + }); + + it('names the event limit and the seq of the stop marker', async () => { + const maxEntries = 20; + const eventLog = await EventLog.open(eventLogPath, { maxEntries }); + try { + for (let index = 0; index < maxEntries; index += 1) { + await eventLog.append('output', { data: `line ${String(index)}\r\n` }); + } + + expect(eventLog.getStop()).toEqual({ + reason: 'events', + seq: maxEntries - EVENT_LOG_TAIL_RESERVE_ENTRIES, + }); + } finally { + await eventLog.close(); + } + }); + + it('reports the stop as soon as the append that stops the log returns, before its marker is written', async () => { + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + await eventLog.append('output', { data: 'prompt$ ' }); + + // Not awaited: the host marks the stop in memory here, and the write + // of the marker settles later. A wait that times out in between must + // still see the stop. + const stopping = eventLog.append('output', { data: 'x'.repeat(6000) }); + expect(eventLog.getStop()).toEqual({ reason: 'bytes', seq: 1 }); + + await stopping; + } finally { + await eventLog.close(); + } + }); + + it('reports the stop of a log that it opened stopped', async () => { + const first = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + await first.append('output', { data: 'prompt$ ' }); + await first.append('output', { data: 'x'.repeat(6000) }); + } finally { + await first.close(); + } + + const reopened = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + expect(reopened.getStop()).toEqual({ reason: 'bytes', seq: 1 }); + } finally { + await reopened.close(); + } + }); + + it('keeps honouring a marker with the reserved prefix that names no limit, and reports an unknown reason', async () => { + // A file that agent-tty did not write: the host has always treated the + // reserved prefix as the stop marker. + await writeFile( + eventLogPath, + `${JSON.stringify({ + seq: 0, + ts: '2026-03-19T12:00:00.000Z', + type: 'output', + payload: { data: 'hello' }, + })}\n${JSON.stringify({ + seq: 1, + ts: '2026-03-19T12:00:00.000Z', + type: 'marker', + payload: { label: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (soon)` }, + })}\n`, + 'utf8', + ); + + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + expect(eventLog.getStop()).toEqual({ reason: 'unknown', seq: 1 }); + // The log starts stopped: it stores nothing more. + await expect(eventLog.append('output', { data: 'later' })).resolves.toBe( + 1, + ); + expect(eventLog.getEvents()).toHaveLength(2); + } finally { + await eventLog.close(); + } + }); +}); diff --git a/test/unit/host/rpcServer.test.ts b/test/unit/host/rpcServer.test.ts index 6c670f89..0c9b7775 100644 --- a/test/unit/host/rpcServer.test.ts +++ b/test/unit/host/rpcServer.test.ts @@ -370,3 +370,58 @@ describe('RpcServer large success responses', () => { }, ); }); + +describe('RpcServer waitForRender results', () => { + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), 'agent-tty-rpc-server-')); + }); + + afterEach(async () => { + await rm(tempDir, { recursive: true, force: true }); + tempDir = ''; + }); + + it('carries the event log stop of a timed-out wait to the client', async () => { + const socketFile = join(tempDir, 'rpc.sock'); + const timedOut = { + matched: false, + timedOut: true, + capturedAtSeq: 7, + eventLogStop: { reason: 'bytes', seq: 7 }, + }; + const server = new RpcServer(socketFile, { + waitForRender: () => Promise.resolve(timedOut), + }); + await server.listen(); + + try { + await expect( + sendRpc(socketFile, 'waitForRender', { text: 'later' }, 5_000), + ).resolves.toEqual(timedOut); + } finally { + await server.close(); + } + }); + + it('refuses a wait result whose event log stop names no reason', async () => { + const socketFile = join(tempDir, 'rpc.sock'); + const server = new RpcServer(socketFile, { + waitForRender: () => + Promise.resolve({ + matched: false, + timedOut: true, + capturedAtSeq: 7, + eventLogStop: { seq: 7 }, + }), + }); + await server.listen(); + + try { + await expect( + sendRpc(socketFile, 'waitForRender', { text: 'later' }, 5_000), + ).rejects.toMatchObject({ code: 'RPC_ERROR' }); + } finally { + await server.close(); + } + }); +}); diff --git a/test/unit/protocol/messages.test.ts b/test/unit/protocol/messages.test.ts index c7c28e60..23e7563b 100644 --- a/test/unit/protocol/messages.test.ts +++ b/test/unit/protocol/messages.test.ts @@ -357,6 +357,31 @@ describe('RPC message schemas', () => { expect(result.success).toBe(true); }); + it('accepts an event log stop with each reason and keeps it minimal', () => { + const inspect = (eventLogStop: unknown) => + InspectResultSchema.safeParse({ + session: createSessionRecord(), + eventCount: 2, + uptime: 1000, + rendererRuntime: { + backend: 'ghostty-web', + mode: 'live-host', + status: 'healthy', + }, + eventLogStop, + }).success; + + expect(inspect({ reason: 'bytes', seq: 0 })).toBe(true); + expect(inspect({ reason: 'events', seq: 249_998 })).toBe(true); + expect(inspect({ reason: 'unknown', seq: 3 })).toBe(true); + expect(inspect({ reason: 'entries', seq: 4 })).toBe(false); + expect(inspect({ reason: 'bytes', seq: -1 })).toBe(false); + expect(inspect({ reason: 'bytes', seq: 1.5 })).toBe(false); + expect(inspect({ reason: 'bytes' })).toBe(false); + expect(inspect({ seq: 4 })).toBe(false); + expect(inspect({ reason: 'bytes', seq: 4, limit: 52_428_800 })).toBe(false); + }); + it('requires rendererRuntime on inspect results', () => { const result = InspectResultSchema.safeParse({ session: createSessionRecord(), @@ -783,6 +808,34 @@ describe('RPC message schemas', () => { ).toBe(true); }); + it('accepts a timed-out waitForRender result that carries the event log stop', () => { + const timedOut = { matched: false, timedOut: true, capturedAtSeq: 7 }; + const withStop = (eventLogStop: unknown) => ({ ...timedOut, eventLogStop }); + + expect(WaitForRenderResultSchema.safeParse(timedOut).success).toBe(true); + for (const reason of ['bytes', 'events', 'unknown']) { + const result = withStop({ reason, seq: 7 }); + expect(WaitForRenderResultSchema.safeParse(result).success).toBe(true); + // The host validates every RPC result against this schema. + expect( + RpcMethodSchemas.waitForRender.result.safeParse(result).success, + ).toBe(true); + } + expect( + WaitForRenderResultSchema.safeParse(withStop({ reason: 'bytes' })) + .success, + ).toBe(false); + expect( + WaitForRenderResultSchema.safeParse(withStop({ reason: 'x', seq: 7 })) + .success, + ).toBe(false); + expect( + WaitForRenderResultSchema.safeParse( + withStop({ reason: 'bytes', seq: 7, limit: 1 }), + ).success, + ).toBe(false); + }); + it('accepts valid record diff results', () => { const side = { sessionId: 'session-01', diff --git a/test/unit/storage/eventLogCodec.test.ts b/test/unit/storage/eventLogCodec.test.ts index e6e5d893..00a36792 100644 --- a/test/unit/storage/eventLogCodec.test.ts +++ b/test/unit/storage/eventLogCodec.test.ts @@ -7,7 +7,11 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import type { EventRecord } from '../../../src/protocol/schemas.js'; import { assertEventLogSize, + EVENT_LOG_LIMIT_LABEL_PREFIX, + eventLogStopLabel, EventLogTooLargeError, + findEventLogStop, + isEventLogStopMarker, MAX_EVENT_LOG_SIZE, parseEventLogContent, readEventLogRecords, @@ -255,3 +259,84 @@ describe('event log codec', () => { }); }); }); + +describe('the stop marker of an event log', () => { + function marker(seq: number, label: string): EventRecord { + return { + seq, + ts: '2026-03-19T12:00:00.000Z', + type: 'marker', + payload: { label }, + }; + } + + it('words the label as the docs quote it, for both limits', () => { + expect(EVENT_LOG_LIMIT_LABEL_PREFIX).toBe( + 'agent-tty: event log limit reached', + ); + expect(eventLogStopLabel('bytes', 52_428_800)).toBe( + 'agent-tty: event log limit reached (52428800 bytes); later events are not recorded', + ); + expect(eventLogStopLabel('events', 250_000)).toBe( + 'agent-tty: event log limit reached (250000 events); later events are not recorded', + ); + }); + + it.each([ + ['bytes', 52_428_800], + ['events', 250_000], + ] as const)( + 'reads back the %s limit that the label names, with the seq of the marker', + (limit, max) => { + expect( + findEventLogStop([ + createEvent(), + marker(1, eventLogStopLabel(limit, max)), + ]), + ).toEqual({ reason: limit, seq: 1 }); + }, + ); + + it('returns undefined when the records hold no stop marker', () => { + expect(findEventLogStop([])).toBeUndefined(); + expect(findEventLogStop(createEvents())).toBeUndefined(); + }); + + it('does not take output or another marker for the stop marker', () => { + const lookAlikes: EventRecord[] = [ + createEvent({ + payload: { data: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (8192 bytes)` }, + }), + marker(1, 'event log limit reached (8192 bytes)'), + marker(2, `note: ${EVENT_LOG_LIMIT_LABEL_PREFIX} (8192 bytes)`), + ]; + + expect(findEventLogStop(lookAlikes)).toBeUndefined(); + expect(lookAlikes.some(isEventLogStopMarker)).toBe(false); + }); + + it('returns the first stop marker when the records hold more than one', () => { + expect( + findEventLogStop([ + marker(0, eventLogStopLabel('bytes', 8192)), + marker(1, eventLogStopLabel('events', 20)), + ]), + ).toEqual({ reason: 'bytes', seq: 0 }); + }); + + it.each([ + ['no limit at all', EVENT_LOG_LIMIT_LABEL_PREFIX], + ['a word for a limit', `${EVENT_LOG_LIMIT_LABEL_PREFIX} (soon)`], + ['an unknown unit', `${EVENT_LOG_LIMIT_LABEL_PREFIX} (12 widgets)`], + ['no number', `${EVENT_LOG_LIMIT_LABEL_PREFIX} (bytes)`], + ['text glued to the prefix', `${EVENT_LOG_LIMIT_LABEL_PREFIX}ed (1 bytes)`], + ])( + 'reports a stop marker that names %s with the reason unknown, and does not throw', + (_case, label) => { + expect(findEventLogStop([createEvent(), marker(1, label)])).toEqual({ + reason: 'unknown', + seq: 1, + }); + }, + ); +}); diff --git a/test/unit/storage/eventLogScan.test.ts b/test/unit/storage/eventLogScan.test.ts new file mode 100644 index 00000000..96c2fefc --- /dev/null +++ b/test/unit/storage/eventLogScan.test.ts @@ -0,0 +1,439 @@ +import { + appendFile, + mkdtemp, + realpath, + rm, + stat, + truncate, + writeFile, +} from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type * as FsPromises from 'node:fs/promises'; + +import { EventLog } from '../../../src/host/eventLog.js'; +import { + EVENT_LOG_LIMIT_LABEL_PREFIX, + eventLogStopLabel, + MAX_EVENT_LOG_SIZE, + readEventLogRecords, + scanEventLog, +} from '../../../src/storage/eventLogCodec.js'; + +// `afterSize` runs right after the scan measures the file, so a test can +// append to the file while the scan is under way, as a live host does. +const hooks = vi.hoisted(() => ({ + afterSize: undefined as undefined | (() => Promise), +})); + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + const measure = handle.stat.bind(handle) as () => Promise; + handle.stat = (async () => { + const size = await measure(); + await hooks.afterSize?.(); + return size; + }) as typeof handle.stat; + return handle; + }, + }; +}); + +let tempDir = ''; +let eventLogPath = ''; + +function recordLine( + seq: number, + type: 'output' | 'marker' | 'exit', + payload: Record, +): string { + return `${JSON.stringify({ + seq, + ts: '2026-03-19T12:00:00.000Z', + type, + payload, + })}\n`; +} + +const stopLine = (seq: number, limit: 'bytes' | 'events' = 'bytes'): string => + recordLine(seq, 'marker', { + label: eventLogStopLabel(limit, limit === 'bytes' ? 8192 : 20), + }); + +describe('scanEventLog', () => { + const LIMIT = 8192; + + beforeEach(async () => { + hooks.afterSize = undefined; + // oxfmt-ignore + tempDir = await realpath(await mkdtemp(join(tmpdir(), 'agent-tty-event-log-scan-'))); + eventLogPath = join(tempDir, 'events.jsonl'); + }); + + afterEach(async () => { + hooks.afterSize = undefined; + if (tempDir.length > 0) { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + /** Stops a log with a real writer, then ends it with the exit record. */ + async function writeStoppedLog( + options: { maxBytes: number } | { maxEntries: number }, + ): Promise { + await writeFile(eventLogPath, '', 'utf8'); + const eventLog = await EventLog.open(eventLogPath, options); + try { + for (let index = 0; index < 1000; index += 1) { + await eventLog.append('output', { data: `line ${String(index)}\r\n` }); + if (eventLog.getStop() !== undefined) { + break; + } + } + await eventLog.append('exit', { exitCode: 0, exitSignal: null }); + } finally { + await eventLog.close(); + } + } + + /** Whether `EventLog.open` keeps the file at `eventLogPath` stopped. */ + async function reopenStays(): Promise<'stopped' | 'open'> { + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + const before = eventLog.getEvents().length; + await eventLog.append('output', { data: 'after the reopen' }); + return eventLog.getEvents().length === before ? 'stopped' : 'open'; + } finally { + await eventLog.close(); + } + } + + it('returns undefined when the event log file is missing', async () => { + await expect(scanEventLog(eventLogPath)).resolves.toBeUndefined(); + }); + + it('returns nothing found for an empty event log file', async () => { + await writeFile(eventLogPath, '', 'utf8'); + + await expect(scanEventLog(eventLogPath)).resolves.toEqual({ + entryCount: 0, + byteCount: 0, + stop: undefined, + }); + }); + + it('counts non-blank lines, valid or not, and the bytes of the file', async () => { + const content = [ + JSON.stringify({ seq: 0, type: 'output' }), + '', + ' ', + JSON.stringify({ seq: 1, type: 'resize' }), + ].join('\n'); + await writeFile(eventLogPath, content, 'utf8'); + + await expect(scanEventLog(eventLogPath)).resolves.toEqual({ + entryCount: 2, + byteCount: Buffer.byteLength(content, 'utf8'), + stop: undefined, + }); + }); + + it('counts the bytes of the file, not its characters', async () => { + const content = recordLine(0, 'output', { data: 'héllo ✓ 😀' }); + await writeFile(eventLogPath, content, 'utf8'); + + const scan = await scanEventLog(eventLogPath); + expect(scan?.byteCount).toBe((await stat(eventLogPath)).size); + expect(scan?.byteCount).toBeGreaterThan(content.length); + }); + + it('rejects empty file paths and rethrows other open errors', async () => { + await expect(scanEventLog('')).rejects.toThrow(); + + // A path whose parent is a regular file raises ENOTDIR, which must + // propagate: only ENOENT means that there is no log. + const blocker = join(tempDir, 'blocker'); + await writeFile(blocker, 'x', 'utf8'); + await expect( + scanEventLog(join(blocker, 'events.jsonl')), + ).rejects.toMatchObject({ code: 'ENOTDIR' }); + }); + + it('returns the byte limit that the writer named, with the seq of its marker', async () => { + await writeStoppedLog({ maxBytes: LIMIT }); + + const records = await readEventLogRecords(eventLogPath); + const marker = records.at(-2); + expect(marker?.type).toBe('marker'); + const scan = await scanEventLog(eventLogPath); + expect(scan).toMatchObject({ + entryCount: records.length, + stop: { reason: 'bytes', seq: marker?.seq }, + }); + }); + + it('returns the event limit that the writer named', async () => { + await writeStoppedLog({ maxEntries: 20 }); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 20, + stop: { reason: 'events', seq: 18 }, + }); + }); + + it('finds nothing in a log that has not stopped', async () => { + const eventLog = await EventLog.open(eventLogPath, { maxBytes: LIMIT }); + try { + await eventLog.append('output', { data: 'hello' }); + await eventLog.append('marker', { label: 'a user marker' }); + await eventLog.append('exit', { exitCode: 0, exitSignal: null }); + } finally { + await eventLog.close(); + } + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 3, + stop: undefined, + }); + }); + + it('does not take output or another marker for the stop marker', async () => { + await writeFile( + eventLogPath, + [ + recordLine(0, 'output', { + data: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (8192 bytes)`, + }), + recordLine(1, 'marker', { + label: 'event log limit reached (8192 bytes)', + }), + recordLine(2, 'marker', { + label: `note: ${EVENT_LOG_LIMIT_LABEL_PREFIX} (8192 bytes)`, + }), + ].join(''), + 'utf8', + ); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + stop: undefined, + }); + }); + + it('returns the first stop marker when a file holds more than one', async () => { + await writeFile( + eventLogPath, + stopLine(0, 'bytes') + stopLine(1, 'events'), + 'utf8', + ); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 2, + stop: { reason: 'bytes', seq: 0 }, + }); + }); + + it('reports a stop marker that names no limit with the reason unknown instead of failing', async () => { + await writeFile( + eventLogPath, + recordLine(0, 'output', { data: 'x' }) + + recordLine(1, 'marker', { + label: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (soon)`, + }), + 'utf8', + ); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + stop: { reason: 'unknown', seq: 1 }, + }); + // The host treats the same marker as a stop when it reopens the log. + expect(await reopenStays()).toBe('stopped'); + }); + + it('reports a stop marker whose limit is not closed with the reason unknown', async () => { + await writeFile( + eventLogPath, + recordLine(0, 'marker', { + label: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (9 bytes`, + }), + 'utf8', + ); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + stop: { reason: 'unknown', seq: 0 }, + }); + }); + + it('reads a log that is over the size limit that replay reads', async () => { + await writeFile(eventLogPath, stopLine(0), 'utf8'); + // Sparse padding: an older version wrote past the limit. + await truncate(eventLogPath, MAX_EVENT_LOG_SIZE + 1024); + + await expect(readEventLogRecords(eventLogPath)).rejects.toThrow( + /exceeds size limit/, + ); + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + byteCount: MAX_EVENT_LOG_SIZE + 1024, + stop: { reason: 'bytes', seq: 0 }, + }); + }); + + describe('agrees with what replay accepts', () => { + const output = (seq: number): string => + recordLine(seq, 'output', { data: `line ${String(seq)}` }); + + it.each([ + ['a line that is not JSON', `${output(0)}not json\n${stopLine(2)}`], + [ + 'a line that is not JSON, with a marker whose seq follows', + `${output(0)}not json\n${stopLine(1)}`, + ], + [ + 'a line that is not an event record', + `${output(0)}{"seq":1,"type":"nope"}\n${stopLine(2)}`, + ], + ['a gap in the seq values', `${output(0)}${stopLine(2)}`], + ['a repeated seq', `${output(0)}${output(0)}${stopLine(2)}`], + ['a first seq that is not 0', `${output(1)}${stopLine(2)}`], + ])('reports no stop for %s before the marker', async (_case, content) => { + await writeFile(eventLogPath, content, 'utf8'); + + // Replay rejects this log, so a stop after the bad line is not one + // that a replay reaches. + await expect(readEventLogRecords(eventLogPath)).rejects.toThrow(); + const scan = await scanEventLog(eventLogPath); + expect(scan?.stop).toBeUndefined(); + // The lines are still counted. + expect(scan?.entryCount).toBe(content.split('\n').length - 1); + }); + + it.each(['\u2028', '\u2029'])( + 'reads an output that holds %j as one record, as replay does', + async (separator) => { + await writeFile( + eventLogPath, + `${recordLine(0, 'output', { data: `before${separator}after` })}${stopLine(1)}`, + 'utf8', + ); + + await expect(readEventLogRecords(eventLogPath)).resolves.toHaveLength( + 2, + ); + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 2, + stop: { reason: 'bytes', seq: 1 }, + }); + }, + ); + + it.each([ + ['a line longer than one read chunk', 'x', 200_000], + ['a 3-byte character on every read chunk boundary', '\u20ac', 70_000], + ['a 4-byte character on every read chunk boundary', '\u{1f600}', 70_000], + ])('reads %s as one record', async (_case, unit, count) => { + const data = unit.repeat(count); + await writeFile( + eventLogPath, + `${recordLine(0, 'output', { data })}${stopLine(1)}`, + 'utf8', + ); + + const records = await readEventLogRecords(eventLogPath); + expect(records[0]?.payload).toEqual({ data }); + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 2, + stop: { reason: 'bytes', seq: 1 }, + }); + }); + + it('reports the stop of a log that replay accepts', async () => { + await writeFile(eventLogPath, `${output(0)}${output(1)}${stopLine(2)}`); + + await expect(readEventLogRecords(eventLogPath)).resolves.toHaveLength(3); + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + stop: { reason: 'bytes', seq: 2 }, + }); + }); + + it('reports a stop that is followed by a bad line, such as one that is still being written', async () => { + const marker = stopLine(1); + await writeFile( + eventLogPath, + `${output(0)}${marker}${marker.slice(0, 30)}`, + 'utf8', + ); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 3, + stop: { reason: 'bytes', seq: 1 }, + }); + }); + + it('reports no stop for a last line that the host has not finished writing', async () => { + const marker = stopLine(1); + // Cut inside the label, so the half line still holds the prefix. + const half = marker.slice( + 0, + marker.indexOf(EVENT_LOG_LIMIT_LABEL_PREFIX) + + EVENT_LOG_LIMIT_LABEL_PREFIX.length + + 4, + ); + await writeFile(eventLogPath, `${output(0)}${half}`, 'utf8'); + + await expect(scanEventLog(eventLogPath)).resolves.toMatchObject({ + entryCount: 2, + stop: undefined, + }); + }); + + it('agrees with EventLog.open on whether a log has stopped', async () => { + // A stopped log: both see the stop. + await writeStoppedLog({ maxBytes: LIMIT }); + expect((await scanEventLog(eventLogPath))?.stop).not.toBeUndefined(); + expect(await reopenStays()).toBe('stopped'); + + // A log with look-alikes only: neither sees a stop. + await writeFile( + eventLogPath, + recordLine(0, 'output', { + data: `${EVENT_LOG_LIMIT_LABEL_PREFIX} (8192 bytes)`, + }) + + recordLine(1, 'marker', { + label: 'event log limit reached (8192 bytes)', + }), + 'utf8', + ); + expect((await scanEventLog(eventLogPath))?.stop).toBeUndefined(); + expect(await reopenStays()).toBe('open'); + }); + }); + + it('describes one snapshot of the file, so records that a live host appends meanwhile are in none of its results', async () => { + const before = `${recordLine(0, 'output', { data: 'x' })}${recordLine(1, 'output', { data: 'y' })}`; + await writeFile(eventLogPath, before, 'utf8'); + // The host writes its stop marker right after the scan measured the file. + hooks.afterSize = async () => { + hooks.afterSize = undefined; + await appendFile(eventLogPath, stopLine(2), 'utf8'); + }; + + await expect(scanEventLog(eventLogPath)).resolves.toEqual({ + entryCount: 2, + byteCount: Buffer.byteLength(before, 'utf8'), + stop: undefined, + }); + // The next scan sees the marker, as all three results together. + const next = await scanEventLog(eventLogPath); + expect(next).toMatchObject({ + entryCount: 3, + stop: { reason: 'bytes', seq: 2 }, + }); + expect(next?.byteCount).toBe((await stat(eventLogPath)).size); + }); +});