-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathstate.ts
More file actions
198 lines (179 loc) · 8.09 KB
/
Copy pathstate.ts
File metadata and controls
198 lines (179 loc) · 8.09 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
import { mkdir, writeFile, readFile, rename } from "node:fs/promises";
import { dirname, join } from "node:path";
import { type } from "arktype";
import { sessionDir } from "./index.js";
import { clearActiveRun, getTestWriteGate, isCrashed } from "./active-run.js";
import { COMMAND_NAME } from "../branding.js";
const ConnectedMcpServerSchema = type({
name: "string",
toolCount: "number",
});
export type ConnectedMcpServer = typeof ConnectedMcpServerSchema.infer;
const RunStateSchema = type({
status: "'running' | 'done' | 'failed' | 'cancelled' | 'crashed'",
turnsUsed: "number",
task: "string",
startedAt: "number",
"finishedAt?": "number",
"error?": "string",
// The resolved "provider:model" identity in use when this record was written.
// Absent only for records predating this field or written outside the run
// lifecycle (e.g. a bare rename of a session with no prior state).
"model?": "string",
// MCP servers connected during the session, with the tool count each
// contributed. Empty until the first server finishes connecting.
"mcpServers?": ConnectedMcpServerSchema.array(),
});
export type RunState = typeof RunStateSchema.infer;
function statePath(cwd: string, sessionId: string, home?: string): string {
return join(sessionDir(cwd, sessionId, home), "run.json");
}
let tmpWriteCounter = 0;
// Write atomically: serialize to a unique temp file, then rename into place so a
// crash mid-write never leaves torn JSON. The temp name combines the pid with a
// monotonic counter so concurrent or rapid successive saves within one process
// never collide on the same temp path (pid alone is not unique per call).
export async function atomicWrite(path: string, content: string): Promise<void> {
await mkdir(dirname(path), { recursive: true });
const tmp = `${path}.${process.pid}.${(tmpWriteCounter += 1)}.tmp`;
await writeFile(tmp, content);
await rename(tmp, path);
}
// A corrupt or shape-invalid state file means resume is silently starting over
// and prior progress is being discarded. Surface it rather than swallowing it.
export function warnUnreadableState(path: string, reason: string): void {
process.stderr.write(`${COMMAND_NAME}: ignoring unreadable state at ${path} (${reason}); starting fresh\n`);
}
// Concurrent saveState calls for the same session (a straggler progress
// snapshot racing a terminal finalize write) have no ordering guarantee
// between their underlying rename()s — the later call could still finish
// first and resurrect a closed run.json as "running". Chaining each session's
// writes onto the previous one forces them to apply in call order, so a
// write issued after another always lands after it regardless of how long
// either write's fs calls take. Keyed by sessionId, not path, since callers
// only ever address one file per session.
const writeChains = new Map<string, Promise<void>>();
// Checked right before a chained write actually fires (not at saveState()
// call time) so a snapshot write still queued behind another one, at the
// moment the crash handler flips this flag, sees it and no-ops instead of
// landing after (and clobbering) the crash write issued via saveCrashState.
// This cannot recall a write whose writeFile/rename has already been
// dispatched to the kernel — that residual window is one atomicWrite call
// wide (a small local JSON write), not the remaining lifetime of the process.
async function atomicWriteUnlessCrashed(path: string, content: string): Promise<void> {
// No-op in production; lets a test hold this write open past the moment
// isCrashed() flips, so the check below is proven rather than assumed.
const gate = getTestWriteGate();
if (gate !== null) await gate;
if (isCrashed()) return;
await atomicWrite(path, content);
}
export async function saveState(
cwd: string,
sessionId: string,
state: RunState,
home?: string,
): Promise<void> {
const path = statePath(cwd, sessionId, home);
const content = JSON.stringify(state, null, 2);
const previous = writeChains.get(sessionId) ?? Promise.resolve();
const write = previous.then(
() => atomicWriteUnlessCrashed(path, content),
() => atomicWriteUnlessCrashed(path, content),
);
// Swallow the error in the chain tail (not in `write`, which still rejects
// for this caller) so one failed save doesn't permanently wedge later
// saves for the same session.
const tail = write.catch(() => {});
writeChains.set(sessionId, tail);
// Once this is the last write for the session, drop the entry so a
// long-lived process doesn't retain a chain per session forever.
void tail.then(() => {
if (writeChains.get(sessionId) === tail) writeChains.delete(sessionId);
});
return write;
}
// Single write path for a terminal RunState: pairs the on-disk status with
// clearing the in-memory active-run handle (active-run.ts) so the two facts
// are set together instead of at two independent call sites that could drift.
// Callers writing a non-terminal ("running") snapshot should call saveState
// directly — clearing the active-run handle on a running snapshot would be
// wrong, not merely redundant.
//
// The clear happens before the saveState await, not after: this run is
// closing out regardless of whether the write below succeeds, and a signal
// or uncaught exception landing during that await must see the handle
// already gone, or it races a second "crashed" write (src/index.ts's process
// handlers, via saveCrashState) against the terminal write in flight here.
// Clearing after the await leaves that exact window open on every terminal
// write, not only the crash path's own.
export async function finalizeRunState(
cwd: string,
sessionId: string,
state: RunState,
home?: string,
): Promise<void> {
clearActiveRun();
await saveState(cwd, sessionId, state, home);
}
// Crash-time terminal write. Deliberately bypasses writeChains: a hung or
// still-pending write for this session (possibly the very write mid-flight
// when the process crashed) must never be awaited here, or a queued write
// that never settles would block the crash handler's process.exit forever.
// Callers must call markCrashed() (src/session/active-run.ts) before this, so
// any snapshot write still queued behind another one in the chain steps
// aside instead of racing this write's rename().
//
// This is a second terminal write path alongside finalizeRunState, and stays
// separate on purpose: its only callers are index.ts's process-level
// uncaughtException/unhandledRejection and signal handlers, reached when a
// crash escapes runTUI's own try/catch entirely. finalizeRunState routes
// through saveState's per-session write chain so writes apply in call order;
// that chain is exactly what a crash exit cannot afford to wait on, since
// process.exit must happen deterministically and a stuck earlier write
// (possibly the one that caused the crash) would otherwise hang it.
export async function saveCrashState(
cwd: string,
sessionId: string,
state: RunState,
home?: string,
): Promise<void> {
const path = statePath(cwd, sessionId, home);
await atomicWrite(path, JSON.stringify(state, null, 2));
}
// Returns the parsed state, or the arktype error summary when the shape is
// invalid, so callers can surface a specific reason rather than "invalid shape".
function parseRunState(data: unknown): RunState | { error: string } {
const result = RunStateSchema(data);
return result instanceof type.errors ? { error: result.summary } : result;
}
export async function loadState(
cwd: string,
sessionId: string,
home?: string,
): Promise<RunState | null> {
const path = statePath(cwd, sessionId, home);
try {
const raw = await readFile(path, "utf8");
const parsed = parseRunState(JSON.parse(raw));
if ("error" in parsed) {
warnUnreadableState(path, `invalid shape: ${parsed.error}`);
return null;
}
return parsed;
} catch (err) {
if (err instanceof SyntaxError) {
warnUnreadableState(path, "corrupt JSON");
return null;
}
if (
typeof err === "object" &&
err !== null &&
"code" in err &&
(err as { code?: unknown }).code === "ENOENT"
) {
return null;
}
throw err;
}
}