feat(session-persistence): expose readRaw for per-session artifacts
The persistence contract gains a concrete readRaw default (undefined for backends without a per-session artifact) and the JSONL backend overrides it with the decode of its physical zstd frames, so a consumer can read the stored artifact text verbatim — the session-log export depends on it.
This commit is contained in:
@@ -18,7 +18,8 @@ import {
|
||||
DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
|
||||
type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type StoredPrefix,
|
||||
type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type SessionRawArtifact,
|
||||
type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
@@ -233,6 +234,61 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a session's stored artifact text verbatim: the durable file bytes
|
||||
* decoded from this backend's physical encoding (complete zstd frames
|
||||
* concatenated, or UTF-8 plaintext). The content is the exact JSONL text the
|
||||
* backend wrote — never a reconstruction from parsed events — so packed-
|
||||
* chunk rows, key order, and line breaks survive byte-for-byte. A torn
|
||||
* final frame is omitted, matching the committed-prefix semantics of every
|
||||
* other read.
|
||||
* @param id - the persisted session to read.
|
||||
* @param signal - optional cancellation for the stat/read/decode work.
|
||||
* @returns the raw artifact text plus the header parsed from its own first
|
||||
* line, or `undefined` when the session has no stored artifact.
|
||||
*/
|
||||
override async readRaw(id: SessionId, signal?: AbortSignal): Promise<SessionRawArtifact | undefined> {
|
||||
signal?.throwIfAborted()
|
||||
await this.ensureRootEncoding()
|
||||
signal?.throwIfAborted()
|
||||
const path = await this.findLog(id, signal)
|
||||
if (path === undefined) return undefined
|
||||
let buffer: Buffer
|
||||
// Revision-stable read: a writer appending between stat and readFile
|
||||
// would yield a torn physical file (see readPrefix).
|
||||
for (;;) {
|
||||
signal?.throwIfAborted()
|
||||
const before = fileRevision(await stat(path, { bigint: true }))
|
||||
buffer = await readFile(path, { signal })
|
||||
signal?.throwIfAborted()
|
||||
const after = fileRevision(await stat(path, { bigint: true }))
|
||||
if (before === after) break
|
||||
}
|
||||
let content: string
|
||||
if (this.compression === 'zstd') {
|
||||
const { frames } = scanZstdFrames(buffer)
|
||||
if (frames.length === 0) return undefined
|
||||
const decoder = createZstdFrameDecoder()
|
||||
const plaintexts: Buffer[] = []
|
||||
// The decoder yields views into a reused buffer; copy each frame's
|
||||
// plaintext immediately so a later concat cannot read overwritten memory.
|
||||
for (const plaintext of decoder.decode(buffer, frames)) {
|
||||
signal?.throwIfAborted()
|
||||
plaintexts.push(Buffer.from(plaintext))
|
||||
}
|
||||
content = Buffer.concat(plaintexts).toString('utf8')
|
||||
} else {
|
||||
content = buffer.toString('utf8')
|
||||
}
|
||||
const meta = parseHeaderMeta(content.split('\n', 1)[0] as string)
|
||||
if (meta === undefined || meta.id !== id) {
|
||||
throw new Error(`corrupt session log: invalid header line in "${path}"`)
|
||||
}
|
||||
// The logical artifact name is `session.jsonl` regardless of the physical
|
||||
// encoding suffix (`.jsonl.zstd` marks compression only).
|
||||
return { meta, filename: 'session.jsonl', content }
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a stored prefix and convert torn-tail state to the opaque marker the
|
||||
* coordinator can round-trip without knowing the physical encoding.
|
||||
|
||||
@@ -217,6 +217,26 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
|
||||
})
|
||||
|
||||
it('readRaw returns the stored artifact text verbatim with its original filename', async () => {
|
||||
const m = meta('raw-read', '/work')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
const raw = await ctx.sessionPersistence.readRaw(m.id)
|
||||
expect(raw).toBeDefined()
|
||||
expect(raw!.filename).toBe('session.jsonl')
|
||||
expect(raw!.meta.id).toBe(m.id)
|
||||
// Byte-identical to the physical file — never a reconstruction.
|
||||
expect(raw!.content).toBe(await readFile(rawLogPath(root, '/work', m.id), 'utf8'))
|
||||
expect(raw!.content.split('\n')[0]).toBe(JSON.stringify(toHeaderLine(m)))
|
||||
const scanned = scanLog(Buffer.from(raw!.content))
|
||||
expect(scanned.events.map(event => event.type)).toEqual(oneTurnLog().map(event => event.type))
|
||||
})
|
||||
|
||||
it('readRaw is undefined for an absent session', async () => {
|
||||
const m = meta('raw-missing', '/work')
|
||||
expect(await ctx.sessionPersistence.readRaw(m.id)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('keeps the same location on resume and gives a fork its own location', async () => {
|
||||
const parent = meta('location-parent', '/work')
|
||||
const parentLocation = ctx.sessionPersistence.locate(parent)
|
||||
|
||||
@@ -356,6 +356,27 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
|
||||
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
|
||||
})
|
||||
|
||||
it('readRaw decodes the compressed artifact back to the original JSONL text', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = await mount(root)
|
||||
const header = meta('raw-read-zstd', '/work')
|
||||
await ctx.sessionPersistence.create(header)
|
||||
await ctx.sessionPersistence.append(header.id, oneTurnLog())
|
||||
|
||||
const raw = await ctx.sessionPersistence.readRaw(header.id)
|
||||
expect(raw).toBeDefined()
|
||||
// The logical name drops the physical encoding suffix.
|
||||
expect(raw!.filename).toBe('session.jsonl')
|
||||
expect(raw!.meta.id).toBe(header.id)
|
||||
expect(raw!.content).toBe([
|
||||
JSON.stringify(toHeaderLine(header)),
|
||||
...oneTurnLog().map(e => JSON.stringify(e)),
|
||||
'',
|
||||
].join('\n'))
|
||||
const scanned = scanLog(Buffer.from(raw!.content))
|
||||
expect(scanned.events.map(event => event.type)).toEqual(oneTurnLog().map(event => event.type))
|
||||
})
|
||||
|
||||
it('resolves the default when a programmatic wrapper bypasses Loader schema normalization', async () => {
|
||||
const root = await freshRoot()
|
||||
const ctx = new Context()
|
||||
|
||||
@@ -30,6 +30,16 @@ export interface SessionInspection {
|
||||
readonly events: readonly SessionEvent[]
|
||||
}
|
||||
|
||||
/** A backend's own raw artifact text for one session, verbatim. */
|
||||
export interface SessionRawArtifact {
|
||||
/** The session header parsed from the artifact's own first line. */
|
||||
readonly meta: SessionHeader
|
||||
/** The artifact's base filename on disk, without any physical encoding suffix. */
|
||||
readonly filename: string
|
||||
/** The artifact's full text content, decoded from the backend's physical encoding. */
|
||||
readonly content: string
|
||||
}
|
||||
|
||||
// The backend-agnostic write-path orchestration first-party backends compose.
|
||||
export {
|
||||
DEFAULT_PREPARED_SESSION_CACHE_SIZE,
|
||||
@@ -83,6 +93,24 @@ export abstract class SessionPersistence extends Service {
|
||||
*/
|
||||
abstract locate(meta: SessionHeader): SessionLocation | undefined
|
||||
|
||||
/**
|
||||
* Read a session's backend-owned artifact text verbatim — the exact durable
|
||||
* bytes the backend wrote (decoded from its physical encoding, e.g. a
|
||||
* decompressed JSONL). The returned `content` is the raw text, not a
|
||||
* reconstruction from parsed events, so it preserves backend-specific
|
||||
* serialization (chunk packing, key order, line breaks). Backends without a
|
||||
* per-session artifact (SQLite) inherit the `undefined` default.
|
||||
* @param _id - the persisted session to read (unused by the default: no
|
||||
* per-session artifact).
|
||||
* @param signal - optional cancellation for backend read work.
|
||||
* @returns the raw artifact plus its parsed header, or `undefined` when the
|
||||
* session is absent or the backend owns no per-session artifact.
|
||||
*/
|
||||
readRaw(_id: SessionId, signal?: AbortSignal): Promise<SessionRawArtifact | undefined> {
|
||||
signal?.throwIfAborted()
|
||||
return Promise.resolve(undefined)
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a new session's metadata. A backend MAY defer the physical write
|
||||
* until the first {@link append} (lazy materialization), in which case a
|
||||
|
||||
Reference in New Issue
Block a user