docs: rebalance prose cleanup and add trimming skill
This commit is contained in:
@@ -75,6 +75,9 @@ function isHeaderLine(value: unknown): value is HeaderLine {
|
||||
* Encode an arbitrary string as a single safe path segment, injectively over ALL JS (UTF-16)
|
||||
* strings — including lone surrogates. A {@link SessionId} is an unvalidated branded string,
|
||||
* so this neutralizes `../`, absolute paths, NUL, and separators before any filesystem use.
|
||||
* Safe code units remain literal; every other unit, including `~`, becomes
|
||||
* `~XXXX`. Operating on code units preserves lone surrogates, while special-
|
||||
* casing `.` and `..` prevents traversal by an otherwise safe whole segment.
|
||||
*
|
||||
* @param raw - the string to encode; must be non-empty (throws on `''`).
|
||||
* @returns the escaped single path segment, decodable back to `raw`.
|
||||
@@ -132,9 +135,10 @@ export function eventLine(event: SessionEvent): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line 0). Returns the
|
||||
* longest prefix of complete, seq-contiguous events plus the byte offset of the end of the
|
||||
* last preserved line (`committedBytes`).
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line
|
||||
* 0). Fully written events in an interrupted final turn remain part of the
|
||||
* prefix. The first unparsable record or seq gap after the last `turn/end`
|
||||
* marks a tolerated torn tail; the same hole in the committed region rejects.
|
||||
*
|
||||
* @param buffer - the raw bytes of the log file (header line first).
|
||||
* @returns the header, the preserved event prefix, and `committedBytes` — the
|
||||
@@ -142,9 +146,8 @@ export function eventLine(event: SessionEvent): string {
|
||||
*/
|
||||
export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionEvent[]; committedBytes: number } {
|
||||
const text = buffer.toString('utf8')
|
||||
// Split into complete (newline-terminated) lines, tracking the byte offset of each line's end
|
||||
// so the truncation point is exact (multi-byte chars make the char offset differ from the
|
||||
// byte offset).
|
||||
// Track complete lines by byte offset: a non-newline tail is torn and ignored,
|
||||
// and a running counter avoids rescanning a long multi-byte log.
|
||||
const lines: { text: string; endByte: number }[] = []
|
||||
let start = 0
|
||||
let byteOffset = 0
|
||||
@@ -172,8 +175,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
|
||||
}
|
||||
const headerLine = parsedHeader
|
||||
|
||||
// Find the committed region: the prefix up to and including the LAST complete `turn/end` in
|
||||
// the WHOLE log.
|
||||
// Parse every complete record first so the last valid `turn/end` determines
|
||||
// whether an earlier hole is committed corruption or an uncommitted tail.
|
||||
interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
|
||||
const parsed: Parsed[] = eventEntries.map((entry) => {
|
||||
try {
|
||||
@@ -191,8 +194,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
|
||||
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
|
||||
}
|
||||
|
||||
// Walk the longest PREFIX of complete, seq-contiguous, parseable event lines (line i is a
|
||||
// parsed event with seq === i).
|
||||
// Preserve the contiguous prefix, including a complete interrupted turn;
|
||||
// holes through the last committed boundary throw, while later holes stop.
|
||||
const preserved: SessionEvent[] = []
|
||||
for (let i = 0; i < parsed.length; i++) {
|
||||
const p = parsed[i]
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
/**
|
||||
* JSONL durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-jsonl`).
|
||||
* JSONL durable session-persistence backend. It stores a header and contiguous
|
||||
* events in one append-only file per session, and delegates orchestration to
|
||||
* {@link PersistenceCoordinator}.
|
||||
* @module @deepseek-ai/dsh-session-persistence-jsonl
|
||||
*/
|
||||
|
||||
@@ -66,7 +68,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
|
||||
constructor(ctx: Context, public config: Config) {
|
||||
super(ctx)
|
||||
// Resolve the configured root to an ABSOLUTE path ONCE, here.
|
||||
// Resolve once so later process.cwd() changes cannot split one backend across roots.
|
||||
this.root = resolve(config.root)
|
||||
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
|
||||
}
|
||||
@@ -85,8 +87,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
return this.coordinator.load(id)
|
||||
}
|
||||
|
||||
// `list` is BOTH the public service method and the PersistenceBackend hook — one method, the
|
||||
// bucket walk below.
|
||||
// One method serves both public `list` and the backend hook; delegating it to
|
||||
// the coordinator would call this hook recursively.
|
||||
|
||||
/**
|
||||
* The per-session init promises, exposed for white-box tests that await a
|
||||
@@ -181,8 +183,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
await mkdir(dir, { recursive: true, mode: 0o700 })
|
||||
await this.syncDir(this.root)
|
||||
const finalPath = logPath(this.root, meta.cwd, meta.id)
|
||||
// Never rename over an existing committed log: materialize is the FIRST write of a session
|
||||
// the backend believes is new.
|
||||
// Materialization is the first write; an existing log is an id collision.
|
||||
/* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
|
||||
if (await this.exists(finalPath)) {
|
||||
throw new Error(`refusing to materialize "${meta.id}": a log already exists on disk (load/resume it instead)`)
|
||||
@@ -207,8 +208,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
await link(tmp, finalPath)
|
||||
linked = true
|
||||
} finally {
|
||||
// If link failed, the temp is the only reference and must be removed before the original
|
||||
// error propagates.
|
||||
// Remove an unpublished temp on failure. After publication, defer cleanup
|
||||
// until the directory entry is durable so cleanup cannot reject a live log.
|
||||
/* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
|
||||
if (!linked) await rm(tmp, { force: true })
|
||||
}
|
||||
@@ -326,7 +327,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
const entries = await readdir(this.root, { withFileTypes: true })
|
||||
return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`)
|
||||
} catch (error) {
|
||||
// ENOENT = the root has not been created yet → genuinely no sessions.
|
||||
// Only an absent root means no sessions; rethrow every other I/O failure.
|
||||
if (isENOENT(error)) return []
|
||||
throw error
|
||||
}
|
||||
|
||||
@@ -54,7 +54,8 @@ runPersistenceContract('jsonl', async () => {
|
||||
}
|
||||
})
|
||||
|
||||
// Run the shared coordinator orchestration suite against the real JSONL backend.
|
||||
// Two mounts share this temp root to exercise reload. `corruptTail` appends a partial,
|
||||
// newline-less fragment past the committed region so coordinator repair runs on real file bytes.
|
||||
runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
|
||||
return {
|
||||
@@ -344,7 +345,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
].join('\n') + '\n'
|
||||
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog PRESERVES the
|
||||
// contiguous prefix (turn/start seq 0) — real interrupted-turn work, not discarded — and
|
||||
// stops at the gap.
|
||||
// stops at the gap. `loadCore`, not this scanner, later closes the orphaned turn.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
@@ -464,7 +465,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('list reads a header line longer than the 8KB read chunk', async () => {
|
||||
// readFirstLine accumulates across reads when the first line exceeds its buffer.
|
||||
// A tolerated extra field makes this valid header exceed the 8192-byte read buffer, proving
|
||||
// `readFirstLine` accumulates chunks before `list()` parses it.
|
||||
const bucket = join(root, '_no-cwd')
|
||||
await mkdir(bucket, { recursive: true })
|
||||
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, pad: 'x'.repeat(9000) })
|
||||
@@ -484,7 +486,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
await sessFiberA.dispose()
|
||||
|
||||
// A NEW live Session object reuses id "reuse".
|
||||
// A new Session object reuses the id. Object-keyed initialization must run independently,
|
||||
// detect the disk collision, and reject instead of appending through session A's stale cursor.
|
||||
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
let b!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
@@ -502,7 +505,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await ctx.sessionPersistence.append(SessionId('x'), oneTurnLog())
|
||||
await ctx.fiber.dispose()
|
||||
|
||||
// Backend 2 over the same root.
|
||||
// Backend 2 creates a no-cwd session whose id exists only in `/w`. Exact `loadLive(id,
|
||||
// undefined)` must not adopt across buckets; the any-cwd collision check then rejects instead
|
||||
// of grafting no-cwd events onto a log with mismatched cwd.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
@@ -570,7 +575,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('list surfaces a non-ENOENT root error (ENOTDIR) instead of reporting no sessions', async () => {
|
||||
// A durable backend must not collapse a storage fault to "no sessions".
|
||||
// A durable backend must not collapse a storage fault to "no sessions". Making the root a
|
||||
// regular file forces ENOTDIR from `readdir`, which must propagate.
|
||||
const filePath = join(root, 'not-a-dir')
|
||||
await writeFile(filePath, 'x')
|
||||
const ctx2 = new Context()
|
||||
@@ -581,8 +587,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('loadLive surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => {
|
||||
// A non-ENOENT error from the per-id open() must surface, not be collapsed to "not found"
|
||||
// (which would let live-adoption proceed under a false absence assumption).
|
||||
// A non-ENOENT per-id open error must surface rather than become "not found" and permit false
|
||||
// live adoption. Making the cwd bucket a regular file forces ENOTDIR for its child log path.
|
||||
const cwd = '/x'
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
@@ -703,7 +709,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
const session = ctx.sessions.create(SessionId('reject-bad'))
|
||||
// Serializability is enforced at the source: Session.append throws on a BigInt-bearing
|
||||
// event before it enters session.events, so the durable log can never diverge from the live
|
||||
// log.
|
||||
// log. The error therefore surfaces synchronously at append, not later during backend flush.
|
||||
expect(() => {
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never, { surfaceOp: 'append' })
|
||||
}).toThrow(/non-JSON-serializable/)
|
||||
|
||||
Reference in New Issue
Block a user