diff --git a/CHANGELOG.md b/CHANGELOG.md index 848f26d..2ddbf01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ All notable changes to Codex Extras are documented in this file. +## Unreleased + +- Add experimental, opt-in session status icons. Use local Codex IPC for live + work and input/approval waits, and bounded log reads for recorded results. + Keep the original chat icons and metadata-only reads by default. +- Show unanswered non-blocking questions, resolve replies independently, and + discard superseded or abandoned questions, including migrated histories. +- Mark new results in green until opened successfully through Codex Extras; + persist review state and protect newer results from stale clicks. +- Use theme-aware status colors and a small gray dot for reviewed, idle, and + unknown states, preserving label alignment. Document protocol limitations. + ## 1.1.0 - Add a Codex Chats Activity Bar panel with local chat history, automatic diff --git a/README.md b/README.md index b5c5777..98dc614 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,9 @@ conversations ordered by their most recent activity. The panel reads Codex's local session index from `$CODEX_HOME/session_index.jsonl` when `CODEX_HOME` is set, or from `~/.codex/session_index.jsonl` otherwise. -It reads only the small metadata prefix needed to identify each chat's workspace; -transcript contents are not parsed or copied. +By default, it reads only the small metadata prefix needed to identify each +chat's workspace; transcript contents are not parsed or copied. The optional +status-icons mode also reads lifecycle events from a bounded log tail (see Settings). ### New Codex Agent toolbar button @@ -47,6 +48,47 @@ The official Codex extension is declared as a dependency and is installed automa | Setting | Default | Description | | --- | --- | --- | | `codexExtras.showNewAgentButton` | `true` | Shows the New Codex Agent button in the editor toolbar. | +| `codexExtras.showSessionStatusIcons` | `false` | Experimental: replaces conversation icons with live task status and recorded completion events. | + +Enable **Codex Extras: Show Session Status Icons** for a spinner while working, +a green check mark for a new result to review, a stop icon for interrupted turns, +or an error icon for failed turns. A yellow question mark means an unanswered +non-blocking question; the agent may still be working. It disappears after all +questions have matching replies. Idle and unknown states show a small gray dot, +preserving text alignment; their status remains available in the tooltip. +Sending a new ordinary user message supersedes earlier questions, including +older histories migrated into a single turn; +interrupted or failed turns do not keep abandoned questions active. +A bell means Codex is waiting for input or approval. +A clock means activity is unconfirmed, not that the task has completed. +Changing the setting takes effect immediately. + +Opening a green chat through Codex Extras marks that specific result as reviewed +after the open command succeeds and its editor tab exists. Its status icon then +becomes a small gray dot, preserving text alignment. This is saved in VS Code extension storage across reloads. A later +completed turn gets a new green mark; opening an older result cannot consume it. +Failed opens do not mark results as reviewed. Blocking waits take priority over +non-blocking questions, and unanswered questions take priority over other states. + +Status icons use theme-aware colors: blue for working, green for new results, +yellow for questions/waiting, orange for interrupted, red for failed, and muted foreground +for unconfirmed activity. Shapes and tooltip labels also identify +each state, so color is not the only cue. + +This optional mode reads at most the last 256 KiB of each local session log and +caches the result. Only explicit lifecycle events determine status; assistant +messages and file modification times do not imply completion. File system events +refresh recorded results without polling. A spinner requires a live `active` +state from Codex's internal local IPC channel; logs alone never produce a spinner. +Losing the connection or receiving an owner-disconnection event invalidates live +status immediately. There is no inactivity timeout for a confirmed running task. + +The IPC adapter is experimental and was checked against Codex extension +`26.1007.21434` (stream protocol version 11). An incompatible protocol falls back +to unknown/unconfirmed status. This is an observer only: it does not start, +resume, interrupt, or take ownership of tasks. See [status implementation notes](docs/SESSION-STATUS.md). +When disabled, session bodies are not read, no status watcher is registered, +and no IPC connection is opened. ## Install @@ -68,7 +110,10 @@ code --install-extension codex-extras-1.1.0.vsix Codex Extras contains no telemetry or network requests. It reads the local Codex session index to populate the chat list and opens editor panels provided by the -official Codex extension. +official Codex extension. If status icons are enabled, it also reads a bounded +tail of session logs locally to identify lifecycle events and receives local IPC +state snapshots. Only runtime status and unanswered question IDs are retained from snapshots; conversation +messages are discarded and are never displayed, saved, or sent elsewhere. ## Contributing diff --git a/assets/status-neutral.svg b/assets/status-neutral.svg new file mode 100644 index 0000000..ea55330 --- /dev/null +++ b/assets/status-neutral.svg @@ -0,0 +1 @@ + diff --git a/docs/SESSION-STATUS.md b/docs/SESSION-STATUS.md new file mode 100644 index 0000000..d61ff20 --- /dev/null +++ b/docs/SESSION-STATUS.md @@ -0,0 +1,92 @@ +# Experimental session status icons + +`codexExtras.showSessionStatusIcons` is off by default. The original chat icon, +metadata-only reads, and watcher behavior remain the default. + +## Sources of status + +- Live state: Codex's internal cross-window IPC, not a separate app-server process. + A different app-server instance cannot reliably report another instance's work. +- Historical results: at most the trailing 256 KiB of a local JSONL session file, + cached by file identity, size, and modification time. Only `event_msg` lifecycle + records count. Unfinished logs always return unconfirmed activity. +- No status polling or inactivity-based completion. File events refresh log + results; IPC messages update runtime status. Reconnection uses a delayed retry + only after the IPC connection closes. + +The public app-server protocol documents `thread/status/changed` and runtime +thread status: . +The installed IDE extension owns a stdio app-server connection and does not +export a public status subscription for other VS Code extensions. + +## Internal IPC contract observed locally + +Verified with `openai.chatgpt` version `26.1007.21434` on Windows. Sources of +confirmation: installed extension implementation, successful observer handshake, +read-only ownership discovery, and repeated runtime-state snapshots. + +- Windows endpoint: `\\.\pipe\codex-ipc`. +- Unix endpoints: `$CODEX_HOME/ipc/ipc.sock` and the legacy per-user socket under + the operating system temporary directory. Unix paths are derived from the + installed implementation; live connectivity was verified on Windows only. +- Framing: 4-byte little-endian byte length followed by UTF-8 JSON. +- Initialize: `initialize`, version 0; receives a client ID. +- Subscribe/unsubscribe: `thread-stream-following-changed`, version 1, + `{hostId: "local", conversationId, following}`. +- Runtime stream: `thread-stream-state-changed`, version 11. Snapshots contain + `change.conversationState.threadRuntimeStatus`. Patches carry base and new + revisions. Retain runtime status and unresolved question IDs, never message text. +- `active` means working; nonempty `activeFlags` means waiting for input/approval. + `idle` means idle, `systemError` means failed, other states are unconfirmed. +- An owner-disconnection event or lost IPC connection clears live states. +- A revision gap requests a fresh snapshot. Unsupported versions cannot retain + a running indicator. Discovery requests are answered with `canHandle: false`. + +## Questions and reviewed results + +Async questions are `agentMessage` items with `questions`. Their IDs are the JSON +tuple `["request_user_input_async", item.id, questionIndex]`. User and steering +messages carrying `` resolve those exact IDs. +Inspect both `turns` and canonical `turnHistory.history.entitiesByKey`; deduplicate +items and match replies independently of entity order. Retain only unresolved IDs. +Use canonical island entry order (or recorded turn start times), not entity map +insertion order, to expire old questions when a newer ordinary user message appears. +Process messages inside each turn in order as well: older migrated histories may +combine several user follow-ups into a single turn. If rehydration replaced the +original call ID with `item-N`, an answer may resolve it only through a unique, +exact question-title/index match from a later reply in known conversation order. +Stable call IDs never use this fallback, and ambiguous matches remain unresolved. +A structured async reply resolves its IDs without superseding remaining questions. +Questions in interrupted/failed turns are abandoned. If history order cannot be +established, do not guess which prompt supersedes which question. +Question/reply patches request a debounced fresh snapshot; ordinary assistant text +streaming does not trigger snapshot requests. Blocking runtime waits have priority, +then unanswered async questions, then the underlying task state. + +A completion's identity is its rollout path plus turn ID (or lifecycle byte offset +for older logs without a turn ID). Opening a green result through Extras records +that identity in `globalState` under `codexExtras.reviewedResults`, only after +`vscode.openWith` resolves and its local conversation tab exists. Capture the +displayed identity before opening so a newer completion cannot be marked reviewed +by a stale click. Reviewed completions use a small gray dot in a 16 px icon slot to +preserve label alignment; a later completion has +a different identity and is green again. No source transcripts are modified. + +The adapter never becomes an owner and sends no task-control requests. Receiving +snapshots can temporarily decode conversation data in memory; it keeps no messages +and does not write source logs, configuration, or transcripts. + +## Limits + +This is an undocumented protocol and may change with Codex updates. A socket +connection and the last runtime state cannot prove progress if an owner remains +connected but stops responding. IPC disconnections are handled immediately; +silent hangs cannot be ruled out. A missing log result remains unknown rather +than becoming a successful completion. Remote hosts and compressed active logs +are not currently supported for status discovery. + +Tests cover option toggling, log parsing and bounds, incomplete writes, live +snapshots and runtime patches, waiting states, owner and transport disconnection, +revision mismatch, protocol mismatch, frame bounds, observer-only behavior, async +question/reply matching, and persistent review markers including failed opens and +new completions racing with old editor opens. diff --git a/extension.js b/extension.js index aae4c3d..604955e 100644 --- a/extension.js +++ b/extension.js @@ -4,10 +4,13 @@ const fs = require("node:fs"); const { readFile } = require("node:fs/promises"); const path = require("node:path"); const vscode = require("vscode"); +const { SessionStatusReader, watchSessionStatus } = require("./session-status"); +const { LiveSessionStatus } = require("./session-live-status"); const { getSessionIndexPath, getSessionsDirectory, isPathInsideWorkspace, + loadSessionFiles, loadSessionWorkspaces, parseSessionIndex, } = require("./session-index"); @@ -23,20 +26,47 @@ const CODEX_EDITOR_VIEW_TYPE = "chatgpt.conversationEditor"; const CODEX_NEW_PANEL_SCHEME = "openai-codex"; const CODEX_ROUTE_AUTHORITY = "route"; const CODEX_NEW_PANEL_PATH = "/extension/panel/new"; +const REVIEWED_RESULTS_KEY = "codexExtras.reviewedResults"; +const NEUTRAL_STATUS_ICON = { + light: path.join(__dirname, "assets", "status-neutral.svg"), + dark: path.join(__dirname, "assets", "status-neutral.svg"), +}; let panelSequence = 0; let openQueue = Promise.resolve(); +const SESSION_STATUSES = { + working: { icon: "loading~spin", color: "charts.blue", label: "Working (live)" }, + waiting: { icon: "bell", color: "charts.yellow", label: "Waiting for input or approval (live)" }, + question: { icon: "question", color: "charts.yellow", label: "Unanswered non-blocking question (agent may still be working)" }, + idle: { label: "Idle (live)" }, + unconfirmed: { icon: "clock", color: "descriptionForeground", label: "Activity unconfirmed" }, + completed: { icon: "pass", color: "testing.iconPassed", label: "New result to review" }, + reviewed: { label: "Result reviewed" }, + interrupted: { icon: "debug-stop", color: "charts.orange", label: "Interrupted" }, + failed: { icon: "error", color: "errorForeground", label: "Failed" }, +}; + class CodexChatItem extends vscode.TreeItem { constructor(session) { super(session.title, vscode.TreeItemCollapsibleState.None); this.id = session.id; + this.resultKey = session.status === "completed" ? session.resultKey : null; + const status = session.showStatusIcons + ? SESSION_STATUSES[session.status] ?? { + label: "Status unknown", + } + : undefined; this.description = [session.workspaceName, formatUpdatedAt(session.updatedAt)] .filter(Boolean) .join(" · "); this.tooltip = [ session.title, + status ? `Status: ${status.label}` : undefined, + session.status === "unconfirmed" && status + ? "Live status is unavailable. An unfinished log does not confirm the task is still running." + : undefined, session.workspacePath ? `Workspace: ${session.workspacePath}` : undefined, session.updatedAt ? `Updated: ${new Date(session.updatedAt).toLocaleString()}` @@ -44,15 +74,18 @@ class CodexChatItem extends vscode.TreeItem { ] .filter(Boolean) .join("\n"); - this.iconPath = new vscode.ThemeIcon("comment-discussion"); + this.iconPath = status && !status.icon ? NEUTRAL_STATUS_ICON : new vscode.ThemeIcon( + status?.icon ?? "comment-discussion", + status ? new vscode.ThemeColor(status.color) : undefined, + ); this.contextValue = "codexChat"; this.command = { command: OPEN_CHAT_COMMAND_ID, title: "Open Codex Chat", - arguments: [session.id], + arguments: [{ id: session.id, resultKey: this.resultKey }], }; this.accessibilityInformation = { - label: `${session.title}, Codex chat`, + label: `${session.title}, Codex chat${status ? `, ${status.label}` : ""}`, }; } } @@ -77,11 +110,15 @@ class CodexChatGroupItem extends vscode.TreeItem { } class CodexChatHistoryProvider { - constructor(sessionIndexPath, sessionsDirectory, getWorkspaceFolders) { + constructor(sessionIndexPath, sessionsDirectory, getWorkspaceFolders, getReviewedResults) { this.sessionIndexPath = sessionIndexPath; this.sessionsDirectory = sessionsDirectory; this.getWorkspaceFolders = getWorkspaceFolders; + this.getReviewedResults = getReviewedResults; + this.displayedResults = new Map(); this.workspaceBySessionId = new Map(); + this.showStatusIcons = false; + this.sessionStatusReader = new SessionStatusReader(); this.changeEmitter = new vscode.EventEmitter(); this.statusEmitter = new vscode.EventEmitter(); this.onDidChangeTreeData = this.changeEmitter.event; @@ -111,9 +148,15 @@ class CodexChatHistoryProvider { const missingWorkspaceIds = sessions .filter((session) => !this.workspaceBySessionId.has(session.id)) .map((session) => session.id); + const showStatusIcons = this.showStatusIcons; + this.liveStatus?.setThreads(sessions.map(({ id }) => id)); + const sessionFiles = showStatusIcons + ? await loadSessionFiles(this.sessionsDirectory, sessions.map(({ id }) => id)) + : undefined; const loadedWorkspaces = await loadSessionWorkspaces( this.sessionsDirectory, missingWorkspaceIds, + sessionFiles, ); for (const [sessionId, workspacePath] of loadedWorkspaces) { @@ -121,10 +164,31 @@ class CodexChatHistoryProvider { } const workspaceFolders = this.getWorkspaceFolders(); - const enrichedSessions = sessions.map((session) => ({ - ...session, - workspacePath: this.workspaceBySessionId.get(session.id), + const enrichedSessions = await Promise.all(sessions.map(async (session) => { + const historicalStatus = sessionFiles?.has(session.id) + ? await this.sessionStatusReader.read(sessionFiles.get(session.id)) + : undefined; + const liveStatus = this.liveStatus?.get(session.id); + const resultKey = sessionFiles?.has(session.id) + ? this.sessionStatusReader.resultKey(sessionFiles.get(session.id)) : undefined; + let status = liveStatus && liveStatus !== "idle" ? liveStatus + : liveStatus === "idle" && historicalStatus === "unconfirmed" ? "idle" + : historicalStatus ?? liveStatus; + if (status === "completed" && resultKey === this.getReviewedResults()[session.id]) { + status = "reviewed"; + } + return { + ...session, + workspacePath: this.workspaceBySessionId.get(session.id), + showStatusIcons, + status, + resultKey, + }; })); + this.displayedResults = new Map(enrichedSessions.map((session) => [ + session.id, session.status === "completed" ? session.resultKey : undefined, + ])); + if (sessionFiles) this.sessionStatusReader.retain(sessionFiles.values()); this.statusEmitter.fire(undefined); return createChatGroups(enrichedSessions, workspaceFolders); @@ -140,6 +204,7 @@ class CodexChatHistoryProvider { } dispose() { + this.sessionStatusReader.clear(); this.changeEmitter.dispose(); this.statusEmitter.dispose(); } @@ -350,6 +415,8 @@ function watchSessionIndex(sessionIndexPath, onChange) { * @param {import("vscode").ExtensionContext} context */ function activate(context) { + let reviewedResults = context.globalState.get(REVIEWED_RESULTS_KEY, {}); + let reviewQueue = Promise.resolve(); const openNewChat = () => { openQueue = openQueue.catch(() => undefined).then(openNewCodexAgent); return openQueue; @@ -359,7 +426,28 @@ function activate(context) { sessionIndexPath, getSessionsDirectory(), () => vscode.workspace.workspaceFolders ?? [], + () => reviewedResults, ); + const openChat = async (session) => { + const id = typeof session === "string" ? session : session?.id; + // Capture the displayed result before opening: a newer turn may finish + // while the editor is being created and must remain unread. + const resultKey = session && typeof session === "object" && "resultKey" in session + ? session.resultKey : historyProvider.displayedResults.get(id); + await openCodexChat(session); + if (!resultKey || !vscode.window.tabGroups.all.some((group) => group.tabs.some((tab) => { + const uri = tab.input?.uri; + return uri?.scheme === CODEX_NEW_PANEL_SCHEME && uri.authority === CODEX_ROUTE_AUTHORITY && + uri.path === `/local/${id}`; + }))) return; + reviewQueue = reviewQueue.catch(() => {}).then(async () => { + const next = { ...reviewedResults, [id]: resultKey }; + await context.globalState.update(REVIEWED_RESULTS_KEY, next); + reviewedResults = next; + historyProvider.refresh(); + }); + await reviewQueue; + }; const historyView = vscode.window.createTreeView(CHAT_HISTORY_VIEW_ID, { treeDataProvider: historyProvider, showCollapseAll: false, @@ -377,11 +465,35 @@ function activate(context) { const workspaceSubscription = vscode.workspace.onDidChangeWorkspaceFolders( () => historyProvider.refresh(), ); + let sessionStatusWatcher; + const updateStatusIcons = () => { + const enabled = vscode.workspace.getConfiguration("codexExtras") + .get("showSessionStatusIcons", false); + sessionStatusWatcher?.dispose(); + historyProvider.liveStatus?.dispose(); + historyProvider.liveStatus = undefined; + sessionStatusWatcher = undefined; + historyProvider.showStatusIcons = enabled; + historyProvider.sessionStatusReader.clear(); + if (enabled) { + historyProvider.liveStatus = new LiveSessionStatus({ + codexHome: path.dirname(sessionIndexPath), onChange: () => historyProvider.refresh(), + }); + sessionStatusWatcher = watchSessionStatus( + historyProvider.sessionsDirectory, () => historyProvider.refresh(), vscode, + ); + } + historyProvider.refresh(); + }; + const configurationSubscription = vscode.workspace.onDidChangeConfiguration((event) => { + if (event.affectsConfiguration("codexExtras.showSessionStatusIcons")) updateStatusIcons(); + }); + updateStatusIcons(); context.subscriptions.push( vscode.commands.registerCommand(COMMAND_ID, openNewChat), vscode.commands.registerCommand(NEW_CHAT_COMMAND_ID, openNewChat), - vscode.commands.registerCommand(OPEN_CHAT_COMMAND_ID, openCodexChat), + vscode.commands.registerCommand(OPEN_CHAT_COMMAND_ID, openChat), vscode.commands.registerCommand(REFRESH_CHATS_COMMAND_ID, () => historyProvider.refresh(), ), @@ -390,6 +502,11 @@ function activate(context) { statusSubscription, visibilitySubscription, workspaceSubscription, + configurationSubscription, + { dispose() { + sessionStatusWatcher?.dispose(); + historyProvider.liveStatus?.dispose(); + } }, watchSessionIndex(sessionIndexPath, () => historyProvider.refresh()), ); } diff --git a/package.json b/package.json index 72f4ffb..1973aef 100644 --- a/package.json +++ b/package.json @@ -110,6 +110,11 @@ "configuration": { "title": "Codex Extras", "properties": { + "codexExtras.showSessionStatusIcons": { + "type": "boolean", + "default": false, + "description": "Experimental: replace chat icons with live task status via Codex's internal local IPC protocol, plus recorded completion/interruption events. Show a spinner only when live status confirms activity; otherwise show unknown/unconfirmed. Disabled by default." + }, "codexExtras.showNewAgentButton": { "type": "boolean", "default": true, diff --git a/scripts/validate.mjs b/scripts/validate.mjs index 3e95090..89e4566 100644 --- a/scripts/validate.mjs +++ b/scripts/validate.mjs @@ -16,6 +16,11 @@ assert.equal(manifest.license, "MIT"); assert.equal(manifest.repository.url, "https://github.com/netroforge/codex-extras.git"); assert.ok(manifest.extensionDependencies.includes("openai.chatgpt")); assert.equal(manifest.main, "./extension.js"); +assert.equal( + manifest.contributes.configuration.properties["codexExtras.showSessionStatusIcons"].default, + false, + "Status icons must remain opt-in", +); assert.ok( manifest.activationEvents.includes("onCommand:codexExtras.newCodexAgent"), ); diff --git a/session-index.js b/session-index.js index b3dc8d9..18ed497 100644 --- a/session-index.js +++ b/session-index.js @@ -129,16 +129,13 @@ async function findSessionFiles(directory, requestedSessionIds, results) { requestedSessionIds.has(sessionId) && !results.has(sessionId) ) { - const workspacePath = await readSessionWorkspace(entryPath); - if (workspacePath) { - results.set(sessionId, workspacePath); - } + results.set(sessionId, entryPath); } }), ); } -async function loadSessionWorkspaces(sessionsDirectory, sessionIds) { +async function loadSessionFiles(sessionsDirectory, sessionIds) { const requestedSessionIds = new Set(sessionIds); const results = new Map(); @@ -149,6 +146,18 @@ async function loadSessionWorkspaces(sessionsDirectory, sessionIds) { return results; } +async function loadSessionWorkspaces(sessionsDirectory, sessionIds, sessionFiles) { + const files = sessionFiles ?? await loadSessionFiles(sessionsDirectory, sessionIds); + const results = new Map(); + await Promise.all(sessionIds.map(async (id) => { + const filePath = files.get(id); + if (!filePath) return; + const workspacePath = await readSessionWorkspace(filePath); + if (workspacePath) results.set(id, workspacePath); + })); + return results; +} + function isPathInsideWorkspace(candidatePath, workspacePath) { if (!candidatePath || !workspacePath) { return false; @@ -171,6 +180,7 @@ module.exports = { getSessionIndexPath, getSessionsDirectory, isPathInsideWorkspace, + loadSessionFiles, loadSessionWorkspaces, parseSessionIndex, }; diff --git a/session-live-status.js b/session-live-status.js new file mode 100644 index 0000000..b5ede52 --- /dev/null +++ b/session-live-status.js @@ -0,0 +1,235 @@ +"use strict"; + +const net = require("node:net"); +const os = require("node:os"); +const path = require("node:path"); +const { randomUUID } = require("node:crypto"); +const { pendingQuestionIds, questionPatchMayChange } = require("./session-questions"); + +// Experimental: this is Codex's internal cross-window protocol, not its public +// app-server API. Reject unknown stream versions rather than infer activity. +const STREAM_VERSION = 11; +const MAX_FRAME_BYTES = 32 * 1024 * 1024; + +function runtimeStatus(value) { + if (value?.type === "active") { + return value.activeFlags?.length ? "waiting" : "working"; + } + if (value?.type === "idle") return "idle"; + if (value?.type === "systemError") return "failed"; + return "unconfirmed"; +} + +class LiveSessionStatus { + constructor({ codexHome, onChange, connect = net.connect }) { + this.onChange = onChange; + this.connect = connect; + this.endpoint = process.platform === "win32" + ? "\\\\.\\pipe\\codex-ipc" + : path.join(codexHome, "ipc", "ipc.sock"); + this.legacyEndpoint = process.platform === "win32" ? undefined + : path.join(os.tmpdir(), "codex-ipc", process.getuid?.() ? `ipc-${process.getuid()}.sock` : "ipc.sock"); + this.threads = new Set(); + this.states = new Map(); + this.snapshotRequests = new Map(); + this.buffer = Buffer.alloc(0); + this.disposed = false; + this.start(); + } + + start(useLegacy = false) { + if (this.disposed) return; + this.clientId = undefined; + this.buffer = Buffer.alloc(0); + const socket = this.connect(useLegacy ? this.legacyEndpoint : this.endpoint); + this.socket = socket; + socket.on("connect", () => { + this.initializeId = randomUUID(); + this.send({ type: "request", method: "initialize", version: 0, + requestId: this.initializeId, sourceClientId: "initializing-client", + params: { clientType: "codex-extras" } }); + }); + socket.on("data", (data) => { + try { + this.buffer = Buffer.concat([this.buffer, data]); + while (this.buffer.length >= 4) { + const length = this.buffer.readUInt32LE(0); + if (!length || length > MAX_FRAME_BYTES) throw new Error("Invalid IPC frame"); + if (this.buffer.length < length + 4) break; + const message = JSON.parse(this.buffer.subarray(4, length + 4)); + this.buffer = this.buffer.length === length + 4 + ? Buffer.alloc(0) : Buffer.from(this.buffer.subarray(length + 4)); + this.receive(message); + } + } catch { + socket.destroy(); + } + }); + socket.on("error", () => {}); + socket.on("close", () => { + if (this.socket !== socket) return; + this.clientId = undefined; + this.buffer = Buffer.alloc(0); + this.states.clear(); + this.clearSnapshotRequests(); + this.onChange(); + if (!this.disposed) { + this.retry = setTimeout(() => this.start(!useLegacy && !!this.legacyEndpoint), 2000); + this.retry.unref(); + } + }); + } + + send(message) { + if (!this.socket?.writable) return; + const body = Buffer.from(JSON.stringify(message)); + const header = Buffer.alloc(4); + header.writeUInt32LE(body.length); + this.socket.write(Buffer.concat([header, body])); + } + + follow(id, following) { + if (!this.clientId) return; + this.send({ type: "broadcast", method: "thread-stream-following-changed", + version: 1, sourceClientId: this.clientId, + params: { hostId: "local", conversationId: id, following } }); + } + + setThreads(ids) { + const next = new Set(ids); + for (const id of this.threads) { + if (!next.has(id)) { + this.follow(id, false); + this.states.delete(id); + clearTimeout(this.snapshotRequests.get(id)); + this.snapshotRequests.delete(id); + } + } + for (const id of next) if (!this.threads.has(id)) this.follow(id, true); + this.threads = next; + } + + requestSnapshot(id) { + if (this.snapshotRequests.has(id)) return; + const timer = setTimeout(() => { + this.snapshotRequests.delete(id); + if (this.threads.has(id)) this.follow(id, true); + }, 100); + timer.unref(); + this.snapshotRequests.set(id, timer); + } + + clearSnapshotRequests() { + for (const timer of this.snapshotRequests.values()) clearTimeout(timer); + this.snapshotRequests.clear(); + } + + receive(message) { + if (message.type === "response" && message.requestId === this.initializeId) { + if (message.resultType !== "success" || typeof message.result?.clientId !== "string") { + this.socket.destroy(); + return; + } + this.clientId = message.result.clientId; + for (const id of this.threads) this.follow(id, true); + return; + } + // This client observes state only. Never accept ownership or execution requests. + if (message.type === "client-discovery-request") { + this.send({ type: "client-discovery-response", requestId: message.requestId, + response: { canHandle: false } }); + return; + } + if (message.type !== "broadcast") return; + if (message.targetClientIds && !message.targetClientIds.includes(this.clientId)) return; + if (message.method === "client-status-changed") { + if (message.params?.status === "disconnected") { + for (const [id, state] of this.states) { + if (state.owner === message.params.clientId) { + this.states.delete(id); + clearTimeout(this.snapshotRequests.get(id)); + this.snapshotRequests.delete(id); + } + } + this.onChange(); + } + return; + } + // An existing owner asks followers to announce themselves after reconnecting. + if (message.method === "thread-stream-following-status-requested") { + if (message.params?.hostId === "local" && this.threads.has(message.params.conversationId)) { + this.follow(message.params.conversationId, true); + } + return; + } + if (message.method !== "thread-stream-state-changed") return; + const { conversationId: id, hostId, change } = message.params ?? {}; + if (hostId !== "local" || !this.threads.has(id)) return; + const previous = this.states.get(id); + if (message.version !== STREAM_VERSION) { + this.states.delete(id); + this.onChange(); + return; + } + let value; + let pendingQuestions = previous?.pendingQuestions ?? new Set(); + if (change?.type === "snapshot") { + // Retain runtime status and unresolved IDs, never question/answer text. + value = change.conversationState?.threadRuntimeStatus; + pendingQuestions = pendingQuestionIds(change.conversationState); + clearTimeout(this.snapshotRequests.get(id)); + this.snapshotRequests.delete(id); + } else if (change?.type === "patches") { + if (!previous || previous.owner !== message.sourceClientId || + previous.revision !== change.baseRevision) { + this.states.delete(id); + this.follow(id, true); + this.onChange(); + return; + } + value = previous.runtime; + for (const patch of change.patches ?? []) { + const segments = Array.isArray(patch.path) ? patch.path + : typeof patch.path === "string" ? patch.path.split("/").slice(1) : []; + if (questionPatchMayChange(patch, segments)) this.requestSnapshot(id); + if (!segments.length) { + value = patch.value?.threadRuntimeStatus; + } else if (segments[0] === "threadRuntimeStatus") { + if (segments.length === 1) value = patch.op === "remove" ? undefined : patch.value; + else if (segments.length === 2 && segments[1] === "type") value = { ...value, type: patch.value }; + else if (segments[1] === "activeFlags") { + // Request a fresh snapshot for nested flag updates; no history patches needed. + value = undefined; + this.follow(id, true); + } + } + } + } else { + this.states.delete(id); + this.onChange(); + return; + } + const baseStatus = runtimeStatus(value); + const status = baseStatus === "waiting" ? "waiting" + : pendingQuestions.size ? "question" : baseStatus; + this.states.set(id, { + owner: message.sourceClientId, revision: change.revision, runtime: value, status, pendingQuestions, + }); + if (status !== previous?.status) this.onChange(); + } + + get(id) { + return this.clientId ? this.states.get(id)?.status : undefined; + } + + dispose() { + this.disposed = true; + clearTimeout(this.retry); + this.clearSnapshotRequests(); + for (const id of this.threads) this.follow(id, false); + this.states.clear(); + this.socket?.destroy(); + } +} + +module.exports = { LiveSessionStatus, runtimeStatus }; diff --git a/session-questions.js b/session-questions.js new file mode 100644 index 0000000..f422ad4 --- /dev/null +++ b/session-questions.js @@ -0,0 +1,130 @@ +"use strict"; + +const REPLY_START = ""; +const REPLY_END = ""; + +function itemQuestionIds(item) { + const asked = []; + const answered = []; + const replies = []; + if (item?.type === "agentMessage" && typeof item.id === "string" && Array.isArray(item.questions)) { + for (let index = 0; index < item.questions.length; index++) { + asked.push(JSON.stringify(["request_user_input_async", item.id, index])); + } + } + if (item?.type === "userMessage" || item?.type === "steeringUserMessage") { + const content = item.type === "userMessage" ? item.content : item.input; + if (!Array.isArray(content)) return { asked, answered, replies }; + const text = content.filter((part) => part?.type === "text") + .map((part) => part.text ?? "").join("\n").trim(); + if (!text.startsWith(REPLY_START) || !text.endsWith(REPLY_END)) return { asked, answered, replies }; + try { + const parsed = JSON.parse(text.slice(REPLY_START.length, -REPLY_END.length)); + for (const reply of Array.isArray(parsed) ? parsed : [parsed]) { + if (typeof reply?.questionItemId !== "string") continue; + let id; + try { id = JSON.parse(reply.questionItemId); } catch { continue; } + if (Array.isArray(id) && id.length === 3 && id[0] === "request_user_input_async" && + typeof id[1] === "string" && Number.isInteger(id[2]) && id[2] >= 0) { + answered.push(JSON.stringify(id)); + replies.push({ id: JSON.stringify(id), index: id[2], title: reply.question }); + } + } + } catch { + // Unsupported or malformed reply markup cannot resolve a question. + } + } + return { asked, answered, replies }; +} + +function orderedTurns(state) { + const history = state?.turnHistory?.history; + const entities = history?.entitiesByKey; + if (entities && Object.keys(entities).length) { + const entries = (history.islands ?? []).flatMap((island) => island.entries ?? []); + if (entries.length) { + // Hydrated entity insertion order is not conversation order. + return { turns: entries.map((entry) => entities[entry.value]).filter(Boolean), ordered: true }; + } + const turns = Object.values(entities); + const ordered = turns.every((turn) => Number.isFinite(turn?.turnStartedAtMs)); + if (ordered) turns.sort((left, right) => left.turnStartedAtMs - right.turnStartedAtMs); + return { turns, ordered }; + } + return { turns: Array.isArray(state?.turns) ? state.turns : [], ordered: true }; +} + +function isNewUserPrompt(item) { + if (item?.type !== "userMessage" || !Array.isArray(item.content)) return false; + const text = item.content.filter((part) => part?.type === "text") + .map((part) => part.text ?? "").join("\n").trim(); + // Even an unsupported/malformed structured reply is not a new task. + return text.length > 0 && !text.startsWith(REPLY_START); +} + +function pendingQuestionIds(state) { + const asked = new Set(); + const answered = new Set(); + const knownIds = new Set(); + const legacy = new Map(); + const replies = []; + let position = 0; + const { turns, ordered } = orderedTurns(state); + for (const turn of turns) { + if (!Array.isArray(turn?.items)) continue; + for (const item of turn.items) { + position++; + if (ordered && isNewUserPrompt(item)) { + asked.clear(); + legacy.clear(); + } + const ids = itemQuestionIds(item); + for (const id of ids.asked) knownIds.add(id); + if (turn.status !== "interrupted" && turn.status !== "failed") { + for (const id of ids.asked) asked.add(id); + if (/^item-\d+$/.test(item.id ?? "")) { + for (let index = 0; index < ids.asked.length; index++) { + legacy.set(ids.asked[index], { title: item.questions[index]?.title, index, position }); + } + } + } + for (const id of ids.answered) answered.add(id); + for (const reply of ids.replies) replies.push({ ...reply, position }); + } + } + for (const id of answered) asked.delete(id); + if (ordered) { + for (const reply of replies) { + if (knownIds.has(reply.id) || typeof reply.title !== "string" || !reply.title) continue; + const candidates = [...legacy].filter(([id, question]) => asked.has(id) && + question.position < reply.position && question.index === reply.index && question.title === reply.title); + // Legacy rehydration can replace a call ID with item-N. Use only a unique, + // exact question-title/index match; never fuzzy-match or inspect answers. + if (candidates.length === 1) asked.delete(candidates[0][0]); + } + } + return asked; +} + +// Refresh a question projection from a snapshot when its source changes. +// Ignore ordinary assistant text deltas; never retain conversation bodies. +function questionPatchMayChange(patch, segments) { + if (!segments.length) return true; + if (segments[0] !== "turnHistory" && segments[0] !== "turns") return false; + if (segments.includes("questions")) return true; + if ((segments.at(-1) === "status" && !segments.includes("items")) || segments.includes("islands")) return true; + if (segments[0] === "turns" && segments.length <= 2) return true; + if (segments[0] === "turnHistory" && segments.length <= 4 && + !["generation", "isComplete", "islands"].includes(segments.at(-1))) return true; + const itemIndex = segments.indexOf("items"); + if (itemIndex < 0) return false; + if (segments.length === itemIndex + 1) return true; + if (segments.length === itemIndex + 2) { + if (patch.op === "remove") return true; + const ids = itemQuestionIds(patch.value); + return ids.asked.length > 0 || ids.answered.length > 0 || isNewUserPrompt(patch.value); + } + return ["input", "content", "type", "id"].includes(segments[itemIndex + 2]); +} + +module.exports = { itemQuestionIds, pendingQuestionIds, questionPatchMayChange }; diff --git a/session-status.js b/session-status.js new file mode 100644 index 0000000..42591f7 --- /dev/null +++ b/session-status.js @@ -0,0 +1,123 @@ +"use strict"; + +const { open } = require("node:fs/promises"); + +const STATUS_TAIL_BYTES = 256 * 1024; +const STATUS_BY_EVENT = new Map([ + ["task_started", "working"], + ["task_complete", "completed"], + ["turn_aborted", "interrupted"], + ["task_failed", "failed"], +]); + +// Only lifecycle events count. An assistant message or a quiet log does not +// mean a turn has ended. Ignore incomplete JSONL records while Codex writes. +function parseSessionStatus(contents, startOffset = 0, previous) { + let result = previous; + let offset = startOffset; + const lines = contents.split("\n"); + for (const line of lines.slice(0, -1)) { + const eventOffset = offset; + offset += Buffer.byteLength(line) + 1; + if (previous && eventOffset <= previous.offset) continue; + let event; + try { + event = JSON.parse(line); + } catch { + continue; + } + if (event.type !== "event_msg") continue; + const status = STATUS_BY_EVENT.get(event.payload?.type); + if (!status) continue; + const turnId = event.payload.turn_id; + if (status !== "working" && result?.status === "working" && + turnId && result.turnId && turnId !== result.turnId) continue; + result = { status, turnId, offset: eventOffset }; + } + return result; +} + +class SessionStatusReader { + constructor() { + this.cache = new Map(); + } + + displayStatus(entry) { + // Logs provide history, not liveness. Only IPC may produce a spinner. + return entry.event?.status === "working" ? "unconfirmed" : entry.event?.status; + } + + async read(filePath) { + let handle; + try { + handle = await open(filePath, "r"); + const stat = await handle.stat(); + const cached = this.cache.get(filePath); + if (cached?.size === stat.size && cached.mtimeMs === stat.mtimeMs && + cached.ino === stat.ino) return this.displayStatus(cached); + const growing = cached && cached.ino === stat.ino && stat.size > cached.size; + const previous = growing ? cached.event : undefined; + const start = Math.max(0, stat.size - STATUS_TAIL_BYTES); + const buffer = Buffer.alloc(stat.size - start); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, start); + // A tail may begin in a JSON string. Skip its first partial record. + const firstLineEnd = start > 0 ? buffer.indexOf(10) : -1; + const skip = start > 0 ? (firstLineEnd < 0 ? bytesRead : firstLineEnd + 1) : 0; + const event = parseSessionStatus( + buffer.toString("utf8", skip, bytesRead), start + skip, previous, + ); + const entry = { size: stat.size, mtimeMs: stat.mtimeMs, ino: stat.ino, event }; + this.cache.set(filePath, entry); + return this.displayStatus(entry); + } catch { + this.cache.delete(filePath); + return undefined; + } finally { + await handle?.close().catch(() => undefined); + } + } + + retain(filePaths) { + const retained = new Set(filePaths); + for (const filePath of this.cache.keys()) { + if (!retained.has(filePath)) { + this.cache.delete(filePath); + } + } + } + + resultKey(filePath) { + const entry = this.cache.get(filePath); + if (entry?.event?.status !== "completed") return undefined; + return JSON.stringify([filePath, entry.event.turnId ?? entry.event.offset]); + } + + clear() { + this.cache.clear(); + } +} + +function watchSessionStatus(directory, onChange, vscode) { + const watcher = vscode.workspace.createFileSystemWatcher( + new vscode.RelativePattern(directory, "**/*.jsonl"), + ); + let timer; + const refresh = () => { + clearTimeout(timer); + timer = setTimeout(onChange, 100); + }; + const subscriptions = [ + watcher.onDidCreate(refresh), + watcher.onDidChange(refresh), + watcher.onDidDelete(refresh), + ]; + return { + dispose() { + clearTimeout(timer); + for (const subscription of subscriptions) subscription.dispose(); + watcher.dispose(); + }, + }; +} + +module.exports = { parseSessionStatus, SessionStatusReader, watchSessionStatus }; diff --git a/test/extension.test.cjs b/test/extension.test.cjs index 24202a6..1ef3981 100644 --- a/test/extension.test.cjs +++ b/test/extension.test.cjs @@ -1,21 +1,25 @@ "use strict"; const assert = require("node:assert/strict"); -const { mkdtemp, mkdir, rm, writeFile } = require("node:fs/promises"); +const { appendFile, mkdtemp, mkdir, rm, writeFile } = require("node:fs/promises"); const Module = require("node:module"); const os = require("node:os"); const path = require("node:path"); const test = require("node:test"); +const { setTimeout: delay } = require("node:timers/promises"); const extensionPath = path.resolve(__dirname, "..", "extension.js"); const originalLoad = Module._load; -function loadExtension(vscode) { +function loadExtension(vscode, LiveStatus) { delete require.cache[extensionPath]; Module._load = function load(request, parent, isMain) { if (request === "vscode") { return vscode; } + if (request === "./session-live-status" && LiveStatus) { + return { LiveSessionStatus: LiveStatus }; + } return originalLoad.call(this, request, parent, isMain); }; @@ -62,7 +66,25 @@ function createHarness(t, tabs = [], options = {}) { } } + const configurationChanged = new EventEmitter(); + let showStatusIcons = options.showSessionStatusIcons ?? false; + const fileWatchers = []; + const liveMonitors = []; + class LiveStatus { + constructor({ onChange }) { + this.states = new Map(); + this.onChange = onChange; + liveMonitors.push(this); + } + setThreads(ids) { this.threads = new Set(ids); } + get(id) { return this.states.get(id); } + dispose() { this.disposed = true; this.states.clear(); } + } + const vscode = { + RelativePattern: class RelativePattern { + constructor(base, pattern) { this.base = base; this.pattern = pattern; } + }, commands: { registerCommand(command, handler) { commandHandlers.set(command, handler); @@ -70,6 +92,11 @@ function createHarness(t, tabs = [], options = {}) { }, async executeCommand(...args) { calls.push(args); + if (options.beforeOpen && args[0] === "vscode.openWith") await options.beforeOpen(); + if (options.failOpen) throw new Error("Synthetic editor opening failure"); + if (args[0] === "vscode.openWith" && args[1]?.path.startsWith("/local/") && !options.skipTabCreation) { + tabs.push({ input: { uri: args[1] } }); + } }, }, Uri: { @@ -82,6 +109,12 @@ function createHarness(t, tabs = [], options = {}) { }, EventEmitter, ThemeIcon: class ThemeIcon { + constructor(id, color) { + this.id = id; + this.color = color; + } + }, + ThemeColor: class ThemeColor { constructor(id) { this.id = id; } @@ -93,6 +126,53 @@ function createHarness(t, tabs = [], options = {}) { Expanded: 2, }, workspace: { + getConfiguration(section) { + assert.equal(section, "codexExtras"); + return { get: (key, fallback) => key === "showSessionStatusIcons" ? showStatusIcons : fallback }; + }, + onDidChangeConfiguration: configurationChanged.event, + createFileSystemWatcher(pattern) { + const created = new EventEmitter(); + const changed = new EventEmitter(); + const deleted = new EventEmitter(); + // Exercise real filesystem changes in the temporary Codex home. The + // production extension delegates watcher lifecycle to VS Code. + let nativeWatcher; + if (options.codexHome) { + nativeWatcher = require("node:fs").watch( + options.codexHome, { recursive: true }, (event, filename) => { + if (!filename) return; + const fsPath = path.join(options.codexHome, filename); + const relative = path.relative(pattern.base, fsPath); + const matches = pattern.pattern === "*" + ? relative !== "" && !relative.includes(path.sep) && !relative.startsWith("..") + : relative !== "" && !relative.startsWith("..") && !path.isAbsolute(relative); + if (matches) { + const emitter = event === "change" ? changed + : require("node:fs").existsSync(fsPath) ? created : deleted; + emitter.fire({ fsPath }); + } + }, + ); + nativeWatcher.on("error", () => {}); + } + const watcher = { + pattern, + disposed: false, + onDidCreate: created.event, + onDidChange: changed.event, + onDidDelete: deleted.event, + dispose() { + this.disposed = true; + nativeWatcher?.close(); + created.dispose(); + changed.dispose(); + deleted.dispose(); + }, + }; + fileWatchers.push(watcher); + return watcher; + }, workspaceFolders: options.workspaceFolders ?? [ { name: "codex-extras", @@ -124,7 +204,14 @@ function createHarness(t, tabs = [], options = {}) { }, }, }; - const context = { subscriptions: [] }; + const storedState = options.storedState ?? new Map(); + const context = { + subscriptions: [], + globalState: { + get: (key, fallback) => storedState.get(key) ?? fallback, + async update(key, value) { storedState.set(key, value); }, + }, + }; const originalCodexHome = process.env.CODEX_HOME; if (options.codexHome) { @@ -132,7 +219,7 @@ function createHarness(t, tabs = [], options = {}) { } try { - loadExtension(vscode).activate(context); + loadExtension(vscode, LiveStatus).activate(context); } finally { if (originalCodexHome === undefined) { delete process.env.CODEX_HOME; @@ -151,6 +238,15 @@ function createHarness(t, tabs = [], options = {}) { calls, commandHandlers, treeViews, + fileWatchers, + liveMonitors, + storedState, + setStatusIcons(value) { + showStatusIcons = value; + configurationChanged.fire({ + affectsConfiguration: (key) => key === "codexExtras.showSessionStatusIcons", + }); + }, runCommand: (command = "codexExtras.newCodexAgent", ...args) => commandHandlers.get(command)(...args), }; @@ -345,3 +441,160 @@ test("groups current workspace chats before other workspaces", async (t) => { assert.equal(otherChats[0].label, "Other chat"); assert.match(otherChats[0].description, /^other · /); }); + +test("status icons are opt-in, update on live events, and toggle without reloading", async (t) => { + const codexHome = await mkdtemp(path.join(os.tmpdir(), "codex-extras-status-ui-")); + t.after(() => rm(codexHome, { recursive: true, force: true })); + const sessions = path.join(codexHome, "sessions"); + await mkdir(sessions); + const id = "019f9c23-f7ec-72c3-8eb6-0018b17bea4f"; + const filePath = path.join(sessions, `rollout-${id}.jsonl`); + const lifecycle = (type) => JSON.stringify({ + timestamp: new Date().toISOString(), type: "event_msg", payload: { type, turn_id: "one" }, + }) + "\n"; + await writeFile(filePath, '{"type":"session_meta","payload":{"cwd":"/work/current"}}\n' + lifecycle("task_complete")); + await writeFile(path.join(codexHome, "session_index.jsonl"), JSON.stringify({ id, thread_name: "Status test" })); + const harness = createHarness(t, [], { codexHome, workspaceFolders: [] }); + const provider = harness.treeViews[0].options.treeDataProvider; + let reads = 0; + const read = provider.sessionStatusReader.read.bind(provider.sessionStatusReader); + provider.sessionStatusReader.read = (...args) => { reads++; return read(...args); }; + const item = async () => { + const groups = await provider.getChildren(); + return (await provider.getChildren(groups[0]))[0]; + }; + assert.equal((await item()).iconPath.id, "comment-discussion"); + assert.equal((await item()).iconPath.color, undefined); + assert.equal(reads, 0, "disabled mode must not read status logs"); + assert.equal(harness.fileWatchers.length, 0, "no status watchers when disabled"); + assert.equal(harness.liveMonitors.length, 0, "no IPC connection when disabled"); + harness.setStatusIcons(true); + assert.equal((await item()).iconPath.id, "pass"); + assert.equal((await item()).iconPath.color.id, "testing.iconPassed"); + const watcher = harness.fileWatchers.at(-1); + assert.equal(watcher.pattern.base, sessions); + assert.equal(watcher.pattern.pattern, "**/*.jsonl"); + let changes = 0; + provider.onDidChangeTreeData(() => changes++); + const waitForRefresh = async (previous) => { + const deadline = Date.now() + 5000; + while (changes === previous && Date.now() < deadline) await delay(25); + assert.ok(changes > previous, "session log changes should refresh icons"); + }; + let previous = changes; + await appendFile(filePath, lifecycle("task_started")); + await waitForRefresh(previous); + assert.equal((await item()).iconPath.id, "clock", "log starts alone must not spin"); + const live = harness.liveMonitors.at(-1); + live.states.set(id, "working"); + live.onChange(); + assert.equal((await item()).iconPath.id, "loading~spin"); + assert.equal((await item()).iconPath.color.id, "charts.blue"); + assert.match((await item()).accessibilityInformation.label, /Working/); + for (const [state, icon, color] of [ + ["waiting", "bell", "charts.yellow"], + ["question", "question", "charts.yellow"], + ]) { + live.states.set(id, state); + const chat = await item(); + assert.equal(chat.iconPath.id, icon); + assert.equal(chat.iconPath.color.id, color); + } + live.states.set(id, "idle"); + const idle = await item(); + assert.equal(idle.iconPath.light, path.resolve(__dirname, "..", "assets", "status-neutral.svg")); + assert.match(idle.tooltip, /Idle \(live\)/); + live.states.clear(); + live.onChange(); + assert.equal((await item()).iconPath.id, "clock", "losing live status removes spinner"); + assert.equal((await item()).iconPath.color.id, "descriptionForeground"); + previous = changes; + await appendFile(filePath, lifecycle("turn_aborted")); + await waitForRefresh(previous); + assert.equal((await item()).iconPath.id, "debug-stop"); + assert.equal((await item()).iconPath.color.id, "charts.orange"); + await appendFile(filePath, lifecycle("task_failed")); + assert.equal((await item()).iconPath.id, "error"); + assert.equal((await item()).iconPath.color.id, "errorForeground"); + await writeFile(filePath, "{}\n"); + const unknown = await item(); + assert.deepEqual(unknown.iconPath, idle.iconPath, "inactive states preserve the same neutral dot"); + assert.match(unknown.tooltip, /Status unknown/); + harness.setStatusIcons(false); + assert.ok(watcher.disposed); + assert.ok(live.disposed); + const previousReads = reads; + assert.equal((await item()).iconPath.id, "comment-discussion"); + assert.equal(reads, previousReads); + assert.equal(provider.sessionStatusReader.cache.size, 0); +}); + +async function resultFixture(t) { + const codexHome = await mkdtemp(path.join(os.tmpdir(), "codex-extras-reviewed-")); + t.after(() => rm(codexHome, { recursive: true, force: true })); + const id = "019f9c23-f7ec-72c3-8eb6-0018b17bea4f"; + await mkdir(path.join(codexHome, "sessions")); + const filePath = path.join(codexHome, "sessions", `rollout-${id}.jsonl`); + const lifecycle = (type, turnId) => JSON.stringify({ + type: "event_msg", payload: { type, turn_id: turnId }, + }) + "\n"; + await writeFile(filePath, lifecycle("task_complete", "turn-one")); + await writeFile(path.join(codexHome, "session_index.jsonl"), JSON.stringify({ id, thread_name: "Review test" })); + const options = { codexHome, workspaceFolders: [], showSessionStatusIcons: true }; + const item = async (harness) => { + const provider = harness.treeViews[0].options.treeDataProvider; + const groups = await provider.getChildren(); + return (await provider.getChildren(groups[0]))[0]; + }; + return { id, filePath, lifecycle, options, item }; +} + +test("opening a green result hides its icon across reloads; a new result turns green again", async (t) => { + const { id, filePath, lifecycle, options, item } = await resultFixture(t); + const harness = createHarness(t, [], options); + assert.equal((await item(harness)).iconPath.id, "pass"); + await harness.runCommand("codexExtras.openChat", id); + assert.equal((await item(harness)).iconPath.light, path.resolve(__dirname, "..", "assets", "status-neutral.svg")); + assert.match((await item(harness)).tooltip, /Result reviewed/); + const reloaded = createHarness(t, [], { ...options, storedState: harness.storedState }); + assert.equal((await item(reloaded)).iconPath.light, path.resolve(__dirname, "..", "assets", "status-neutral.svg")); + await appendFile(filePath, lifecycle("task_started", "turn-two")); + assert.equal((await item(reloaded)).iconPath.id, "clock"); + await appendFile(filePath, lifecycle("task_complete", "turn-two")); + assert.equal((await item(reloaded)).iconPath.id, "pass"); + await reloaded.runCommand("codexExtras.openChat", { id }); + assert.equal((await item(reloaded)).iconPath.dark, path.resolve(__dirname, "..", "assets", "status-neutral.svg")); +}); + +test("a failed open or missing editor tab cannot consume a result", async (t) => { + const { id, options, item } = await resultFixture(t); + const harness = createHarness(t, [], { ...options, failOpen: true }); + assert.equal((await item(harness)).iconPath.id, "pass"); + await assert.rejects(harness.runCommand("codexExtras.openChat", id), /opening failure/); + assert.equal((await item(harness)).iconPath.id, "pass"); + assert.equal(harness.storedState.size, 0); + const noTab = createHarness(t, [], { ...options, skipTabCreation: true }); + await item(noTab); + await noTab.runCommand("codexExtras.openChat", id); + assert.equal((await item(noTab)).iconPath.id, "pass"); +}); + +test("opening an old result never consumes a newer turn that finishes during opening", async (t) => { + const { id, filePath, lifecycle, options, item } = await resultFixture(t); + const harness = createHarness(t, [], { ...options, async beforeOpen() { + await appendFile(filePath, lifecycle("task_started", "turn-two") + lifecycle("task_complete", "turn-two")); + } }); + await item(harness); + await harness.runCommand("codexExtras.openChat", id); + assert.equal((await item(harness)).iconPath.id, "pass"); +}); + +test("a stale tree item consumes only the result that item displayed", async (t) => { + const { filePath, lifecycle, options, item } = await resultFixture(t); + const harness = createHarness(t, [], options); + const oldItem = await item(harness); + await appendFile(filePath, lifecycle("task_complete", "turn-two")); + await item(harness); + await harness.runCommand(oldItem.command.command, ...oldItem.command.arguments); + assert.equal((await item(harness)).iconPath.id, "pass"); +}); diff --git a/test/session-live-status.test.cjs b/test/session-live-status.test.cjs new file mode 100644 index 0000000..d48a7d5 --- /dev/null +++ b/test/session-live-status.test.cjs @@ -0,0 +1,193 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const { EventEmitter } = require("node:events"); +const test = require("node:test"); +const { LiveSessionStatus, runtimeStatus } = require("../session-live-status"); + +function frame(message) { + const body = Buffer.from(JSON.stringify(message)); + const header = Buffer.alloc(4); + header.writeUInt32LE(body.length); + return Buffer.concat([header, body]); +} + +function harness(t) { + class Socket extends EventEmitter { + writable = true; + writes = []; + write(buffer) { this.writes.push(JSON.parse(buffer.subarray(4))); } + destroy() { + if (!this.writable) return; + this.writable = false; + this.emit("close"); + } + } + const socket = new Socket(); + let changes = 0; + const live = new LiveSessionStatus({ + codexHome: "/tmp/codex-extras-test-home", + onChange: () => changes++, + connect: () => socket, + }); + t.after(() => live.dispose()); + live.setThreads(["thread-one"]); + socket.emit("connect"); + const initialize = socket.writes[0]; + socket.emit("data", frame({ type: "response", requestId: initialize.requestId, + method: "initialize", resultType: "success", result: { clientId: "observer" } })); + const snapshot = (runtime, revision = 1, extra = {}) => ({ + type: "broadcast", method: "thread-stream-state-changed", version: 11, + sourceClientId: "owner", params: { + hostId: "local", conversationId: "thread-one", + change: { type: "snapshot", revision, conversationState: { threadRuntimeStatus: runtime } }, + }, ...extra, + }); + return { live, socket, snapshot, changes: () => changes }; +} + +test("maps runtime activity, approval waits, idle, errors, and unknown states", () => { + assert.equal(runtimeStatus({ type: "active", activeFlags: [] }), "working"); + assert.equal(runtimeStatus({ type: "active", activeFlags: ["waitingOnApproval"] }), "waiting"); + assert.equal(runtimeStatus({ type: "idle" }), "idle"); + assert.equal(runtimeStatus({ type: "systemError" }), "failed"); + assert.equal(runtimeStatus({ type: "notLoaded" }), "unconfirmed"); + assert.equal(runtimeStatus({ type: "future-state" }), "unconfirmed"); +}); + +test("initializes an observer and decodes split and combined IPC frames", (t) => { + const { live, socket, snapshot } = harness(t); + assert.equal(socket.writes[0].method, "initialize"); + assert.deepEqual(socket.writes[1].params, { + hostId: "local", conversationId: "thread-one", following: true, + }); + const message = frame(snapshot({ type: "active", activeFlags: [] })); + socket.emit("data", message.subarray(0, 2)); + assert.equal(live.get("thread-one"), undefined); + socket.emit("data", message.subarray(2)); + assert.equal(live.get("thread-one"), "working"); + socket.emit("data", Buffer.concat([ + frame(snapshot({ type: "idle" }, 2)), + frame(snapshot({ type: "systemError" }, 3)), + ])); + assert.equal(live.get("thread-one"), "failed"); + socket.emit("data", frame({ type: "client-discovery-request", requestId: "discovery" })); + assert.deepEqual(socket.writes.at(-1).response, { canHandle: false }); +}); + +test("owner disconnection and IPC loss immediately invalidate the spinner", (t) => { + const { live, socket, snapshot } = harness(t); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + socket.emit("data", frame({ type: "broadcast", method: "client-status-changed", + params: { clientId: "owner", status: "disconnected" } })); + assert.equal(live.get("thread-one"), undefined); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + assert.equal(live.get("thread-one"), "working"); + socket.destroy(); + assert.equal(live.get("thread-one"), undefined); +}); + +test("applies runtime patches and invalidates gaps or unsupported protocol versions", (t) => { + const { live, socket, snapshot } = harness(t); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + const patch = (baseRevision, revision, patches) => ({ + type: "broadcast", method: "thread-stream-state-changed", version: 11, + sourceClientId: "owner", params: { hostId: "local", conversationId: "thread-one", + change: { type: "patches", baseRevision, revision, patches } }, + }); + socket.emit("data", frame(patch(1, 2, [{ op: "replace", path: ["threadRuntimeStatus"], value: { type: "idle" } }]))); + assert.equal(live.get("thread-one"), "idle"); + socket.emit("data", frame(patch(2, 3, [{ op: "replace", path: "/threadRuntimeStatus", value: { type: "active", activeFlags: [] } }]))); + assert.equal(live.get("thread-one"), "working"); + socket.emit("data", frame(patch(4, 5, []))); + assert.equal(live.get("thread-one"), undefined); + assert.equal(socket.writes.at(-1).params.following, true, "revision gaps request a fresh snapshot"); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }, 2, { version: 999 }))); + assert.equal(live.get("thread-one"), undefined, "unsupported versions cannot retain a spinner"); +}); + +test("ignores messages for other clients, threads, and hosts; retains no transcript", (t) => { + const { live, socket, snapshot } = harness(t); + socket.emit("data", frame(snapshot({ type: "active" }, 1, { targetClientIds: ["someone-else"] }))); + assert.equal(live.get("thread-one"), undefined); + const other = snapshot({ type: "active" }); + other.params.hostId = "remote"; + socket.emit("data", frame(other)); + assert.equal(live.get("thread-one"), undefined); + const valid = snapshot({ type: "active", activeFlags: [] }); + valid.params.change.conversationState.turns = [{ text: "private transcript" }]; + socket.emit("data", frame(valid)); + assert.equal(JSON.stringify([...live.states]).includes("private transcript"), false); + live.setThreads([]); + assert.equal(live.states.size, 0); + assert.equal(socket.writes.at(-1).params.following, false); +}); + +test("rejects oversized frames and closes the connection on dispose", (t) => { + const { live, socket, snapshot } = harness(t); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + const header = Buffer.alloc(4); + header.writeUInt32LE(33 * 1024 * 1024); + socket.emit("data", header); + assert.equal(socket.writable, false); + assert.equal(live.get("thread-one"), undefined); + live.dispose(); + assert.equal(live.states.size, 0); +}); + +test("unanswered async questions override work or idle; blocking waits take priority", (t) => { + const { live, socket, snapshot } = harness(t); + const withQuestions = (runtime, answered = false) => { + const message = snapshot(runtime); + const items = [{ type: "agentMessage", id: "call-one", questions: [{ title: "Private question" }] }]; + if (answered) items.push({ type: "userMessage", content: [{ type: "text", text: + '' + JSON.stringify([{ + questionItemId: JSON.stringify(["request_user_input_async", "call-one", 0]), + answer: "Private answer", + }]) + '', + }] }); + message.params.change.conversationState.turnHistory = { history: { entitiesByKey: { one: { items } } } }; + return message; + }; + socket.emit("data", frame(withQuestions({ type: "active", activeFlags: [] }))); + assert.equal(live.get("thread-one"), "question"); + const retained = [...live.states.values()][0]; + assert.equal(retained.pendingQuestions.size, 1); + assert.equal(JSON.stringify({ ...retained, pendingQuestions: [...retained.pendingQuestions] }).includes("Private"), false); + socket.emit("data", frame(withQuestions({ type: "idle" }))); + assert.equal(live.get("thread-one"), "question", "questions remain visible after agent finishes"); + socket.emit("data", frame(withQuestions({ type: "active", activeFlags: ["waitingOnUserInput"] }))); + assert.equal(live.get("thread-one"), "waiting"); + socket.emit("data", frame(withQuestions({ type: "active", activeFlags: [] }, true))); + assert.equal(live.get("thread-one"), "working"); + socket.emit("data", frame(withQuestions({ type: "idle" }, true))); + assert.equal(live.get("thread-one"), "idle"); +}); + +test("coalesces question patch reconciliation and cancels it on disconnect", (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + const { live, socket, snapshot } = harness(t); + socket.emit("data", frame(snapshot({ type: "active", activeFlags: [] }))); + const before = socket.writes.length; + const questionPath = ["turnHistory", "history", "entitiesByKey", "one", "items", 0, "questions"]; + const patch = (baseRevision, revision, path) => ({ + type: "broadcast", method: "thread-stream-state-changed", version: 11, + sourceClientId: "owner", params: { hostId: "local", conversationId: "thread-one", + change: { type: "patches", baseRevision, revision, patches: [{ op: "add", path, value: [] }] } }, + }); + socket.emit("data", frame(patch(1, 2, questionPath))); + socket.emit("data", frame(patch(2, 3, questionPath))); + assert.equal(socket.writes.length, before); + t.mock.timers.tick(100); + assert.equal(socket.writes.length, before + 1); + assert.equal(socket.writes.at(-1).params.following, true); + socket.emit("data", frame(patch(3, 4, questionPath.slice(0, -1).concat("text")))); + t.mock.timers.tick(100); + assert.equal(socket.writes.length, before + 1, "text streaming does not request snapshots"); + socket.emit("data", frame(patch(4, 5, questionPath))); + socket.destroy(); + t.mock.timers.tick(100); + assert.equal(socket.writes.length, before + 1, "disconnect cancels pending reconciliation"); + assert.equal(live.get("thread-one"), undefined); +}); diff --git a/test/session-questions.test.cjs b/test/session-questions.test.cjs new file mode 100644 index 0000000..8c04768 --- /dev/null +++ b/test/session-questions.test.cjs @@ -0,0 +1,116 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const test = require("node:test"); +const { itemQuestionIds, pendingQuestionIds, questionPatchMayChange } = require("../session-questions"); + +const id = (call, index) => JSON.stringify(["request_user_input_async", call, index]); +const ask = (call, count) => ({ + type: "agentMessage", id: call, + questions: Array.from({ length: count }, () => ({ title: "Synthetic question" })), +}); +const reply = (ids, type = "steeringUserMessage") => ({ + type, + [type === "userMessage" ? "content" : "input"]: [{ type: "text", text: + '\n' + JSON.stringify(ids.map((questionItemId) => ({ + questionItemId, question: "Synthetic question", answer: "Synthetic answer", + }))) + '\n', + }], +}); + +test("tracks each async question independently and clears only exact matching replies", () => { + const state = { turns: [{ items: [ask("call-one", 2), ask("call-two", 1), reply([id("call-one", 0)])] }] }; + assert.deepEqual([...pendingQuestionIds(state)].sort(), [id("call-one", 1), id("call-two", 0)].sort()); + state.turns[0].items.push(reply([id("call-one", 1), id("call-two", 0)], "userMessage")); + assert.equal(pendingQuestionIds(state).size, 0); +}); + +test("reconstructs canonical history regardless of entity order and duplicate items", () => { + const question = ask("call-one", 1); + const state = { + turns: [{ items: [question] }], + turnHistory: { history: { entitiesByKey: { + later: { items: [reply([id("call-one", 0)])] }, + earlier: { items: [question] }, + } } }, + }; + assert.equal(pendingQuestionIds(state).size, 0); +}); + +test("assistant-quoted markup and replies for other IDs cannot resolve questions", () => { + const markup = reply([id("call-one", 0)]).input; + assert.deepEqual(itemQuestionIds({ type: "agentMessage", content: markup }).answered, []); + const state = { turns: [{ items: [ + { type: "userMessage", content: [{ type: "text", text: "I answered elsewhere" }] }, + ask("call-one", 1), + { type: "userInputResponse", questions: [{ id: "blocking-question" }], completed: true }, + reply([id("different-call", 0)]), reply([id("call-one", 5)]), + ] }] }; + assert.deepEqual([...pendingQuestionIds(state)], [id("call-one", 0)]); + assert.deepEqual(itemQuestionIds({ type: "userMessage", content: [{ type: "text", text: + "not JSON", + }] }).answered, []); +}); + +test("legacy rehydrated question IDs match a unique exact question title and index", () => { + assert.equal(pendingQuestionIds({ turns: [{ items: [ + ask("item-3", 1), reply([id("original-call", 0)], "userMessage"), + ] }] }).size, 0); + assert.equal(pendingQuestionIds({ turns: [{ items: [ + ask("stable-call", 1), reply([id("different-call", 0)], "userMessage"), + ] }] }).size, 1, "stable IDs must never use title fallback"); + assert.equal(pendingQuestionIds({ turns: [{ items: [ + ask("item-3", 1), ask("item-4", 1), reply([id("original-call", 0)], "userMessage"), + ] }] }).size, 2, "ambiguous titles must not guess which question was answered"); +}); + +test("a new user prompt supersedes earlier questions even inside a migrated single turn", () => { + const items = [ask("item-3", 1), { type: "userMessage", content: [{ type: "text", text: "New instructions" }] }]; + assert.equal(pendingQuestionIds({ turns: [{ items }] }).size, 0); + items.push(ask("new-call", 1)); + assert.deepEqual([...pendingQuestionIds({ turns: [{ items }] })], [id("new-call", 0)]); +}); + +test("question and reply patches request reconciliation, text streaming does not", () => { + const prefix = ["turnHistory", "history", "entitiesByKey", "turn-one", "items"]; + assert.equal(questionPatchMayChange({ op: "add", value: ask("call-one", 1) }, [...prefix, 3]), true); + assert.equal(questionPatchMayChange({ op: "add", value: reply([id("call-one", 0)]) }, [...prefix, 4]), true); + assert.equal(questionPatchMayChange({ op: "replace" }, [...prefix, 3, "questions"]), true); + assert.equal(questionPatchMayChange({ op: "remove" }, [...prefix, 3]), true); + assert.equal(questionPatchMayChange({ op: "replace" }, [...prefix, 3, "text"]), false); + assert.equal(questionPatchMayChange({ op: "replace" }, ["threadRuntimeStatus"]), false); + assert.equal(questionPatchMayChange({ op: "add", value: { type: "toolCall", id: "tool" } }, [...prefix, 3]), false); + assert.equal(questionPatchMayChange({ op: "add", value: { + type: "userMessage", content: [{ type: "text", text: "New task" }], + } }, [...prefix, 4]), true); +}); + +test("a newer ordinary user turn supersedes old async questions in canonical history order", () => { + const prompt = { type: "userMessage", content: [{ type: "text", text: "Continue with the next task" }] }; + const state = { turnHistory: { history: { + // Entity insertion order differs from conversation order, as in a hydrated UI. + entitiesByKey: { + new: { items: [prompt] }, + old: { items: [ask("old-call", 1)] }, + }, + islands: [{ entries: [{ value: "old" }, { value: "new" }] }], + } } }; + assert.equal(pendingQuestionIds(state).size, 0); + state.turnHistory.history.entitiesByKey.new.items.push(ask("new-call", 1)); + assert.deepEqual([...pendingQuestionIds(state)], [id("new-call", 0)]); +}); + +test("an async reply in a newer turn does not supersede other pending questions", () => { + const state = { turns: [ + { items: [ask("call-one", 2)] }, + { items: [reply([id("call-one", 0)], "userMessage")] }, + ] }; + assert.deepEqual([...pendingQuestionIds(state)], [id("call-one", 1)]); +}); + +test("interrupted and failed turns do not keep abandoned questions active", () => { + for (const status of ["interrupted", "failed"]) { + assert.equal(pendingQuestionIds({ turns: [{ status, items: [ask("old-call", 1)] }] }).size, 0); + } + assert.equal(pendingQuestionIds({ turns: [{ status: "completed", items: [ask("current-call", 1)] }] }).size, 1); +}); diff --git a/test/session-status.test.cjs b/test/session-status.test.cjs new file mode 100644 index 0000000..2280aa9 --- /dev/null +++ b/test/session-status.test.cjs @@ -0,0 +1,82 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const { appendFile, mkdtemp, rm, writeFile } = require("node:fs/promises"); +const os = require("node:os"); +const path = require("node:path"); +const test = require("node:test"); +const { parseSessionStatus, SessionStatusReader } = require("../session-status"); + +const event = (type, turnId = "turn-1") => + JSON.stringify({ timestamp: new Date().toISOString(), type: "event_msg", payload: { type, turn_id: turnId } }) + "\n"; + +test("tracks explicit starts, completion, interruption, and failure across turns", () => { + assert.equal(parseSessionStatus(event("task_started")).status, "working"); + assert.equal(parseSessionStatus(event("task_started") + event("task_complete")).status, "completed"); + assert.equal(parseSessionStatus(event("turn_aborted")).status, "interrupted"); + assert.equal(parseSessionStatus(event("task_failed")).status, "failed"); + assert.equal(parseSessionStatus(event("task_complete") + event("task_started", "turn-2")).status, "working"); + assert.equal(parseSessionStatus(event("task_started", "new") + event("task_complete", "old")).status, "working"); +}); + +test("does not infer completion from messages, tool output, or partial JSONL", () => { + const message = JSON.stringify({ type: "response_item", payload: { type: "message", role: "assistant" } }) + "\n"; + assert.equal(parseSessionStatus(message), undefined); + assert.equal(parseSessionStatus(event("task_started") + message).status, "working"); + assert.equal(parseSessionStatus("not JSON\n" + event("unknown")), undefined); + assert.equal(parseSessionStatus(event("task_started") + event("task_complete").trimEnd()).status, "working"); +}); + +test("reads bounded tails, retains status while growing, resets on truncation and deletion", async (t) => { + const root = await mkdtemp(path.join(os.tmpdir(), "codex-extras-status-")); + t.after(() => rm(root, { recursive: true, force: true })); + const filePath = path.join(root, "session.jsonl"); + const reader = new SessionStatusReader(); + t.after(() => reader.clear()); + assert.equal(await reader.read(filePath), undefined); + await writeFile(filePath, event("task_started")); + assert.equal(await reader.read(filePath), "unconfirmed"); + assert.equal(await reader.read(filePath), "unconfirmed"); + await appendFile(filePath, JSON.stringify({ type: "response_item", text: "x".repeat(300 * 1024) }) + "\n"); + assert.equal(await reader.read(filePath), "unconfirmed", "cached start survives large appended records"); + const freshReader = new SessionStatusReader(); + t.after(() => freshReader.clear()); + assert.equal(await freshReader.read(filePath), undefined, "does not read beyond the tail budget"); + await appendFile(filePath, event("task_complete")); + assert.equal(await reader.read(filePath), "completed"); + assert.equal(await freshReader.read(filePath), "completed"); + await writeFile(filePath, "{}\n"); + assert.equal(await reader.read(filePath), undefined, "old state cannot survive truncation"); + await rm(filePath); + assert.equal(await reader.read(filePath), undefined); + assert.equal(reader.cache.size, 0); +}); + +test("handles partial writes and UTF-8 without interpreting a partial first record", async (t) => { + const root = await mkdtemp(path.join(os.tmpdir(), "codex-extras-status-")); + t.after(() => rm(root, { recursive: true, force: true })); + const filePath = path.join(root, "session.jsonl"); + const reader = new SessionStatusReader(); + t.after(() => reader.clear()); + await writeFile(filePath, event("task_started") + event("task_complete").trimEnd()); + assert.equal(await reader.read(filePath), "unconfirmed"); + await appendFile(filePath, "\n"); + assert.equal(await reader.read(filePath), "completed"); + await writeFile(filePath, JSON.stringify({ text: "ą".repeat(300 * 1024) }) + "\n" + event("turn_aborted")); + assert.equal(await new SessionStatusReader().read(filePath), "interrupted"); + reader.retain([]); + assert.equal(reader.cache.size, 0); +}); + +test("logs alone never show a spinner, even with fresh token events", async (t) => { + const root = await mkdtemp(path.join(os.tmpdir(), "codex-extras-status-")); + t.after(() => rm(root, { recursive: true, force: true })); + const filePath = path.join(root, "session.jsonl"); + const reader = new SessionStatusReader(); + await writeFile(filePath, event("task_started")); + assert.equal(await reader.read(filePath), "unconfirmed"); + await appendFile(filePath, event("token_count")); + assert.equal(await reader.read(filePath), "unconfirmed"); + await appendFile(filePath, event("task_complete")); + assert.equal(await reader.read(filePath), "completed"); +});