diff --git a/.changeset/wire-journal-corruption-recovery.md b/.changeset/wire-journal-corruption-recovery.md new file mode 100644 index 000000000..7026e3da8 --- /dev/null +++ b/.changeset/wire-journal-corruption-recovery.md @@ -0,0 +1,5 @@ +--- +"@pymodel/pythinker-code": patch +--- + +Fix sessions that fail to resume when their session journal is truncated or corrupted, for example after the disk fills up. diff --git a/packages/agent-core-v2/src/app/telemetry/events.ts b/packages/agent-core-v2/src/app/telemetry/events.ts index 3336ef882..db33b08aa 100644 --- a/packages/agent-core-v2/src/app/telemetry/events.ts +++ b/packages/agent-core-v2/src/app/telemetry/events.ts @@ -479,6 +479,13 @@ export interface SessionLoadFailedEvent { reason: string; } +export interface WireRepairEvent { + kind: 'corrupted' | 'truncated'; + outcome: 'repaired' | 'failed'; + dropped_count: number; + backup_created: boolean; +} + export interface FirstLaunchEvent {} export interface ExitEvent { @@ -1058,6 +1065,16 @@ export const telemetryEventDefinitions = { comment: 'A session resume fails.', properties: { reason: 'Error code, error name, or unknown' }, }), + wire_repair: defineTelemetryEvent({ + owner: 'pythinker-code', + comment: 'A corrupted wire journal is truncated to its valid prefix and healed on disk.', + properties: { + kind: 'Corruption kind: unparseable middle line or torn final line', + outcome: 'Whether the on-disk repair succeeded', + dropped_count: 'Journal lines dropped from the corrupted tail', + backup_created: 'Whether a first-time .bak backup of the corrupted file was created', + }, + }), first_launch: defineTelemetryEvent({ owner: 'pythinker-code', comment: 'The CLI runs for the first time on this device.', diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts index 31554bcf6..b4eee62e9 100644 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts @@ -7,6 +7,7 @@ import { AppendLogCorruptedError, IAppendLogStore, type AppendLogOptions, + type AppendLogReadOptions, } from '#/persistence/interface/appendLogStore'; const textEncoder = new TextEncoder(); @@ -48,8 +49,9 @@ export class AppendLogStore implements IAppendLogStore { this.scheduleFlush(scope, key, state); } - async *read(scope: string, key: string): AsyncIterable { + async *read(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable { await this.flushLog(scope, key); + const onTruncate = options?.onTruncate; const textDecoder = new TextDecoder(); let pending = ''; let lineNumber = 0; @@ -60,7 +62,14 @@ export class AppendLogStore implements IAppendLogStore { const raw = pending.slice(0, newlineIndex); pending = pending.slice(newlineIndex + 1); lineNumber++; - const record = this.parseLine(raw, scope, key, lineNumber, false); + let record: R | undefined; + try { + record = this.parseLine(raw, scope, key, lineNumber, false); + } catch (error) { + if (onTruncate === undefined) throw error; + onTruncate({ lineNumber, reason: 'corrupted', cause: error }); + return; + } if (record !== undefined) yield record; newlineIndex = pending.indexOf('\n'); } @@ -69,7 +78,12 @@ export class AppendLogStore implements IAppendLogStore { if (pending.length > 0) { lineNumber++; const record = this.parseLine(pending, scope, key, lineNumber, true); - if (record !== undefined) yield record; + if (record !== undefined) { + yield record; + } else if (onTruncate !== undefined) { + const line = pending.endsWith('\r') ? pending.slice(0, -1) : pending; + if (line.length > 0) onTruncate({ lineNumber, reason: 'truncated' }); + } } } diff --git a/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts b/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts index c8f6b3fca..a1d78ccb9 100644 --- a/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts +++ b/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts @@ -21,11 +21,21 @@ export interface AppendLogOptions { readonly onError?: (error: unknown) => void; } +export interface AppendLogTruncation { + readonly lineNumber: number; + readonly reason: 'corrupted' | 'truncated'; + readonly cause?: unknown; +} + +export interface AppendLogReadOptions { + readonly onTruncate?: (truncation: AppendLogTruncation) => void; +} + export interface IAppendLogStore { readonly _serviceBrand: undefined; append(scope: string, key: string, record: R, options?: AppendLogOptions): void; - read(scope: string, key: string): AsyncIterable; + read(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable; rewrite(scope: string, key: string, records: readonly R[]): Promise; flush(): Promise; close(): Promise; diff --git a/packages/agent-core-v2/src/wire/repair.ts b/packages/agent-core-v2/src/wire/repair.ts new file mode 100644 index 000000000..801793bcf --- /dev/null +++ b/packages/agent-core-v2/src/wire/repair.ts @@ -0,0 +1,77 @@ +import type { ILogService } from '#/_base/log/log'; +import type { ITelemetryService } from '#/app/telemetry/telemetry'; +import type { + AppendLogTruncation, + IAppendLogStore, +} from '#/persistence/interface/appendLogStore'; +import type { IFileSystemStorageService } from '#/persistence/interface/storage'; + +export interface WireJournalRepairServices { + readonly appendLog: IAppendLogStore; + readonly storage: IFileSystemStorageService; + readonly log: ILogService; + readonly telemetry: ITelemetryService; +} + +export function wireJournalBackupKey(key: string): string { + return `${key}.bak`; +} + +export async function repairWireJournal( + services: WireJournalRepairServices, + scope: string, + key: string, + records: readonly unknown[], + truncation: AppendLogTruncation, +): Promise { + const { appendLog, storage, log, telemetry } = services; + let backupCreated = false; + let outcome: 'repaired' | 'failed' = 'repaired'; + let droppedCount = 0; + let repairError: unknown; + try { + const original = await storage.read(scope, key); + if (original !== undefined) { + droppedCount = Math.max(0, countJournalLines(original) - truncation.lineNumber + 1); + const backupKey = wireJournalBackupKey(key); + if ((await storage.size(scope, backupKey)) === undefined) { + await storage.write(scope, backupKey, original, { atomic: true }); + backupCreated = true; + } + } + await appendLog.rewrite(scope, key, records); + } catch (error) { + outcome = 'failed'; + repairError = error; + } + log.warn('corrupted wire journal truncated to its valid prefix', { + scope, + key, + lineNumber: truncation.lineNumber, + reason: truncation.reason, + outcome, + droppedCount, + backupCreated, + error: repairError instanceof Error ? repairError.message : undefined, + }); + telemetry.track2('wire_repair', { + kind: truncation.reason, + outcome, + dropped_count: droppedCount, + backup_created: backupCreated, + }); +} + +function countJournalLines(data: Uint8Array): number { + let lines = 0; + let hasContent = false; + for (const byte of data) { + if (byte === 0x0a) { + lines++; + hasContent = false; + } else { + hasContent = true; + } + } + return hasContent ? lines + 1 : lines; +} diff --git a/packages/agent-core-v2/src/wire/wireService.ts b/packages/agent-core-v2/src/wire/wireService.ts index 5a40900c7..7dbdf75b8 100644 --- a/packages/agent-core-v2/src/wire/wireService.ts +++ b/packages/agent-core-v2/src/wire/wireService.ts @@ -1,15 +1,21 @@ import { onUnexpectedError } from '#/_base/errors/unexpectedError'; import { Service } from '#/_base/di/service'; +import { ILogService } from '#/_base/log/log'; import { LifecycleScope } from '#/app/scopes'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; +import { ITelemetryService } from '#/app/telemetry/telemetry'; import type { ContentPart } from '#/kosong/contract/message'; -import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; -import { StorageError, StorageErrors } from '#/persistence/interface/storage'; +import { + type AppendLogTruncation, + IAppendLogStore, +} from '#/persistence/interface/appendLogStore'; +import { IFileSystemStorageService, StorageError, StorageErrors } from '#/persistence/interface/storage'; import { IWireService } from './wire'; import { WireError, WireErrors } from './errors'; +import { repairWireJournal } from './repair'; import { WIRE_PROTOCOL_VERSION, isNewerWireVersion, @@ -38,6 +44,9 @@ export class WireService extends Service implements IWireService { @IAgentScopeContext scopeContext: IAgentScopeContext, @IAppendLogStore private readonly log: IAppendLogStore, @IAgentBlobService private readonly blobService: IAgentBlobService, + @IFileSystemStorageService private readonly storage: IFileSystemStorageService, + @ILogService private readonly logger: ILogService, + @ITelemetryService private readonly telemetry: ITelemetryService, ) { super(); this.wireScope = scopeContext.scope(); @@ -45,7 +54,8 @@ export class WireService extends Service implements IWireService { } async seal(): Promise { - for await (const record of this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY)) { + const tolerate = { onTruncate: () => {} }; + for await (const record of this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY, tolerate)) { void record; return; } @@ -78,7 +88,12 @@ export class WireService extends Service implements IWireService { } async *readJournal(): AsyncIterable { - const source = this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY); + let truncation: AppendLogTruncation | undefined; + const source = this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY, { + onTruncate: (info) => { + truncation = info; + }, + }); let migrations: readonly WireMigration[] = []; let rewrittenRecords: WireRecord[] | undefined; let newerWireVersion = false; @@ -128,11 +143,42 @@ export class WireService extends Service implements IWireService { if (!hasRecords) { rewrittenRecords = [createWireMetadataRecord()]; } - if (rewrittenRecords !== undefined) { + if (truncation !== undefined) { + await this.repairJournal(truncation, rewrittenRecords); + } else if (rewrittenRecords !== undefined) { await this.log.rewrite(this.wireScope, AGENT_WIRE_RECORD_KEY, rewrittenRecords); } } + private async repairJournal( + truncation: AppendLogTruncation, + rewrittenRecords: WireRecord[] | undefined, + ): Promise { + let records: WireRecord[] = rewrittenRecords ?? []; + if (rewrittenRecords === undefined) { + const tolerate = { onTruncate: () => {} }; + for await (const record of this.log.read( + this.wireScope, + AGENT_WIRE_RECORD_KEY, + tolerate, + )) { + records.push(record); + } + } + await repairWireJournal( + { + appendLog: this.log, + storage: this.storage, + log: this.logger, + telemetry: this.telemetry, + }, + this.wireScope, + AGENT_WIRE_RECORD_KEY, + records, + truncation, + ); + } + async flush(): Promise { await this.persistQueue; await this.log.flush(); diff --git a/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts index b5ed990d8..a6b2ac436 100644 --- a/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts @@ -10,6 +10,7 @@ import { } from '#/_base/di/scope'; import { unwrapErrorCause } from '#/_base/errors/errors'; import { AsyncEmitter, Emitter, type Event, type IWaitUntil } from '#/_base/event'; +import { ILogService } from '#/_base/log/log'; import { drainLogCloses } from '#/_base/log/logService'; import { DEFAULT_PLAN_MODE_SECTION } from '#/features/plan/configSection'; import { IAgentPlanService } from '#/features/plan/plan'; @@ -27,8 +28,12 @@ import { import { ITelemetryService } from '#/app/telemetry/telemetry'; import { ErrorCodes, Error2, isError2 } from '#/errors'; import { IHostFileSystem, type HostDirEntry } from '#/os/interface/hostFileSystem'; -import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { + type AppendLogTruncation, + IAppendLogStore, +} from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IAgentLifecycleService, MAIN_AGENT_ID, @@ -51,6 +56,7 @@ import { createWireMetadataRecord, type WireRecord, } from '#/wire/record'; +import { repairWireJournal } from '#/wire/repair'; import { IModelCatalog } from '#/kosong/model/catalog'; import { IModelService } from '#/kosong/model/model'; import { IProviderService } from '#/kosong/provider/provider'; @@ -145,6 +151,8 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec @ISessionIndexMirror private readonly indexMirror: ISessionIndexMirror, @IAppendLogStore private readonly appendLogStore: IAppendLogStore, @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore, + @IFileSystemStorageService private readonly storage: IFileSystemStorageService, + @ILogService private readonly log: ILogService, @IHostFileSystem private readonly hostFs: IHostFileSystem, @IEventService private readonly event: IEventService, @ITelemetryService private readonly telemetry: ITelemetryService, @@ -663,12 +671,30 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec await agentHandle.accessor.get(IEventDispatcher).flush(); } } - return collect( - this.appendLogStore.read( - agentScopeOf(sessionScopeOf(this.handlerScope, sourceSessionId), agentId), - AGENT_WIRE_RECORD_KEY, - ), + const scope = agentScopeOf(sessionScopeOf(this.handlerScope, sourceSessionId), agentId); + let truncation: AppendLogTruncation | undefined; + const records = await collect( + this.appendLogStore.read(scope, AGENT_WIRE_RECORD_KEY, { + onTruncate: (info) => { + truncation = info; + }, + }), ); + if (truncation !== undefined) { + await repairWireJournal( + { + appendLog: this.appendLogStore, + storage: this.storage, + log: this.log, + telemetry: this.telemetry, + }, + scope, + AGENT_WIRE_RECORD_KEY, + records, + truncation, + ); + } + return records; } private async pruneTruncatedForkFiles( diff --git a/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts b/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts index 053bfc35f..d3ee195fb 100644 --- a/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts +++ b/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts @@ -26,6 +26,7 @@ import { IModelService } from '#/kosong/model/model'; import { IProviderService } from '#/kosong/provider/provider'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { Error2, ErrorCodes } from '#/errors'; import { IHostEnvironment } from '#/os/interface/hostEnvironment'; import { LocalRuntimeProviderFactory } from '#/runtime/localRuntime'; @@ -76,6 +77,7 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { @ITelemetryService private readonly telemetry: ITelemetryService, @IAppendLogStore private readonly appendLogStore: IAppendLogStore, @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore, + @IFileSystemStorageService private readonly storage: IFileSystemStorageService, private readonly unitHostFactory: RuntimeUnitHostFactory = new SharedRuntimeUnitHostFactory(), ) { this.providers.set('local', new LocalRuntimeProviderFactory()); @@ -153,7 +155,7 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { this.instances.delete(workspaceId); const attachments = this.attachments.get(workspaceId); this.attachments.delete(workspaceId); - if (attachments !== undefined) for (const attachment of [...attachments.values()].reverse()) await attachment.dispose(); + if (attachments !== undefined) for (const attachment of [...attachments.values()].toReversed()) await attachment.dispose(); await instance.dispose(); this.changeEmitter.fire({ workspaceId }); } @@ -169,18 +171,18 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { } } catch (error) { this.providers.delete(factory.id); - for (const instance of attached.reverse()) await this.detach(instance.id, factory.id); + for (const instance of attached.toReversed()) await this.detach(instance.id, factory.id); throw error; } return { dispose: async () => { if (this.providers.get(factory.id) !== factory) return; this.providers.delete(factory.id); - for (const workspaceId of [...this.attachments.keys()].reverse()) await this.detach(workspaceId, factory.id); + for (const workspaceId of [...this.attachments.keys()].toReversed()) await this.detach(workspaceId, factory.id); } }; } async dispose(): Promise { - for (const workspaceId of [...this.instances.keys()].reverse()) await this.close(workspaceId); + for (const workspaceId of [...this.instances.keys()].toReversed()) await this.close(workspaceId); this.changeEmitter.dispose(); } @@ -225,6 +227,8 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { this.indexMirror, this.appendLogStore, this.docs, + this.storage, + this.log, input.fs, this.event, this.telemetry, @@ -256,7 +260,7 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { const attachments = this.attachments.get(instance.id); this.attachments.delete(instance.id); if (attachments !== undefined) { - for (const attachment of [...attachments.values()].reverse()) await attachment.dispose(); + for (const attachment of [...attachments.values()].toReversed()) await attachment.dispose(); } await instance.dispose(); throw error; diff --git a/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts b/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts index 3ccc1129a..f8b8c4a72 100644 --- a/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts +++ b/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts @@ -3,7 +3,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; -import { AppendLogCorruptedError, IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { AppendLogCorruptedError, IAppendLogStore, type AppendLogTruncation } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; @@ -634,6 +634,56 @@ describe('AppendLogStore', () => { }); }); + it('stops at a corrupted middle line and reports it when onTruncate is given', async () => { + const raw = `${JSON.stringify({ n: 1 })}\nGARBAGE\n${JSON.stringify({ n: 3 })}\n`; + await storage.append(SCOPE, KEY, enc.encode(raw)); + const truncations: AppendLogTruncation[] = []; + + const out: Rec[] = []; + for await (const r of record.read(SCOPE, KEY, { + onTruncate: (truncation) => truncations.push(truncation), + })) { + out.push(r); + } + + expect(out).toEqual([{ n: 1 }]); + expect(truncations).toHaveLength(1); + expect(truncations[0]).toMatchObject({ lineNumber: 2, reason: 'corrupted' }); + expect(truncations[0]!.cause).toBeInstanceOf(AppendLogCorruptedError); + }); + + it('reports a torn final line as truncation when onTruncate is given', async () => { + const raw = `${JSON.stringify({ n: 1 })}\n${JSON.stringify({ n: 2 }).slice(0, 4)}`; + await storage.append(SCOPE, KEY, enc.encode(raw)); + const truncations: AppendLogTruncation[] = []; + + const out: Rec[] = []; + for await (const r of record.read(SCOPE, KEY, { + onTruncate: (truncation) => truncations.push(truncation), + })) { + out.push(r); + } + + expect(out).toEqual([{ n: 1 }]); + expect(truncations).toEqual([{ lineNumber: 2, reason: 'truncated' }]); + }); + + it('does not report truncation for a clean log when onTruncate is given', async () => { + const raw = `${JSON.stringify({ n: 1 })}\n${JSON.stringify({ n: 2 })}\n`; + await storage.append(SCOPE, KEY, enc.encode(raw)); + const truncations: AppendLogTruncation[] = []; + + const out: Rec[] = []; + for await (const r of record.read(SCOPE, KEY, { + onTruncate: (truncation) => truncations.push(truncation), + })) { + out.push(r); + } + + expect(out).toEqual([{ n: 1 }, { n: 2 }]); + expect(truncations).toEqual([]); + }); + it('reads across chunk boundaries (stream read splits lines)', async () => { const full = `${JSON.stringify({ n: 1 })}\n${JSON.stringify({ n: 2 })}\n${JSON.stringify({ n: 3 })}\n`; const bytes = enc.encode(full); diff --git a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts index 1f4df88af..8c8889402 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -77,6 +77,8 @@ import { IAgentPluginService } from '#/agent/plugin/agentPlugin'; import { ILogService } from '#/_base/log/log'; import { IPluginService } from '#/app/plugin/plugin'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata'; @@ -214,6 +216,7 @@ describe('AgentLifecycleService', () => { ix.get(IAgentStateService).contributeState(permissionModeKey); ix.get(IAgentStateService).contributeState(permissionModeConfiguredKey); ix.stub(IAppendLogStore, recordingAppendLog().store); + ix.stub(IFileSystemStorageService, new InMemoryStorageService()); stubBlobPassThrough(ix); registerAgent = vi.fn().mockResolvedValue(undefined); atomicDocs = new Map(); diff --git a/packages/agent-core-v2/test/wire/persistence.test.ts b/packages/agent-core-v2/test/wire/persistence.test.ts index c32315106..29c1d3c7f 100644 --- a/packages/agent-core-v2/test/wire/persistence.test.ts +++ b/packages/agent-core-v2/test/wire/persistence.test.ts @@ -1,5 +1,5 @@ import { randomBytes } from 'node:crypto'; -import { mkdir, readFile, rm } from 'node:fs/promises'; +import { mkdir, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'pathe'; @@ -16,10 +16,11 @@ import { type WireRecord, } from '#/index'; import { IWireService } from '#/wire/wire'; +import { noopTelemetryService } from '#/app/telemetry/telemetry'; import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; -import { registerTestAgentWire } from './stubs'; +import { noopLogger, registerTestAgentWire } from './stubs'; const cleanups: string[] = []; const disposables: DisposableStore[] = []; @@ -57,22 +58,33 @@ function createAppendLogHarness(storage: IFileSystemStorageService): IAppendLogS return ix.get(IAppendLogStore); } -function createAgentWireHarness(log: IAppendLogStore): IWireService { +function createAgentWireHarness( + log: IAppendLogStore, + storage?: IFileSystemStorageService, +): IWireService { const disposable = new DisposableStore(); disposables.push(disposable); const ix = disposable.add(new TestInstantiationService()); - return registerTestAgentWire(ix, SCOPE, { log }); + return registerTestAgentWire(ix, SCOPE, { + log, + storage, + logger: noopLogger, + telemetry: noopTelemetryService, + }); } async function createFileAppendLogHarness(): Promise<{ readonly dir: string; readonly log: IAppendLogStore; + readonly storage: IFileSystemStorageService; }> { const dir = await makeDir('wire-jsonl-test'); + const storage = new FileStorageService(dir); return { dir, - log: createAppendLogHarness(new FileStorageService(dir)), + log: createAppendLogHarness(storage), + storage, }; } @@ -412,3 +424,58 @@ describe('WireService migration rewrite', () => { await expect(drainJournal(svc)).rejects.toThrow('disk full'); }); }); + +describe('WireService corruption repair on disk', () => { + async function seedCorruptJournal(dir: string, raw: string): Promise { + await mkdir(join(dir, SCOPE), { recursive: true }); + await writeFile(join(dir, SCOPE, KEY), raw); + } + + it('restores from the valid prefix and heals the file with a byte-identical backup', async () => { + const { dir, log, storage } = await createFileAppendLogHarness(); + const metadata = JSON.stringify({ + type: 'metadata', + protocol_version: WIRE_PROTOCOL_VERSION, + created_at: 1, + }); + const kept = JSON.stringify({ type: 'wire.test.kept', time: 2 }); + const raw = `${metadata}\n${kept}\nGARBAGE\n${JSON.stringify({ type: 'wire.test.dropped', time: 3 })}\n`; + await seedCorruptJournal(dir, raw); + const svc = createAgentWireHarness(log, storage); + + const yielded = await drainJournal(svc); + await log.close(); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'wire.test.kept', time: 2 }, + ]); + expect(await readFile(join(dir, SCOPE, KEY), 'utf8')).toBe(`${metadata}\n${kept}\n`); + expect(await readFile(join(dir, SCOPE, `${KEY}.bak`), 'utf8')).toBe(raw); + }); + + it('heals a torn final line left by a disk-full crash', async () => { + const { dir, log, storage } = await createFileAppendLogHarness(); + const metadata = JSON.stringify({ + type: 'metadata', + protocol_version: WIRE_PROTOCOL_VERSION, + created_at: 1, + }); + const kept = JSON.stringify({ type: 'wire.test.kept', time: 2 }); + const torn = JSON.stringify({ type: 'wire.test.torn', time: 3 }).slice(0, 12); + await seedCorruptJournal(dir, `${metadata}\n${kept}\n${torn}`); + const svc = createAgentWireHarness(log, storage); + + const yielded = await drainJournal(svc); + await log.close(); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'wire.test.kept', time: 2 }, + ]); + expect(await readFile(join(dir, SCOPE, KEY), 'utf8')).toBe(`${metadata}\n${kept}\n`); + expect(await readFile(join(dir, SCOPE, `${KEY}.bak`), 'utf8')).toBe( + `${metadata}\n${kept}\n${torn}`, + ); + }); +}); diff --git a/packages/agent-core-v2/test/wire/store-event.test.ts b/packages/agent-core-v2/test/wire/store-event.test.ts index 7302feb0e..0543007c9 100644 --- a/packages/agent-core-v2/test/wire/store-event.test.ts +++ b/packages/agent-core-v2/test/wire/store-event.test.ts @@ -9,18 +9,25 @@ import type { ContentPart } from '#/kosong/contract/message'; import { Event2 } from '#/app/event/event2'; import { IEventBus } from '#/app/event/eventBus'; import { EventBusService } from '#/app/event/eventBusService'; +import { noopTelemetryService } from '#/app/telemetry/telemetry'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { IAgentStateService } from '#/agent/state/agentState'; +import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; +import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; +import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IEventDispatcher } from '#/state/eventDispatcher'; import { EventDispatcherService } from '#/state/eventDispatcherService'; import { defineState } from '#/state/state'; -import type { WireRecord } from '#/wire/record'; +import { WIRE_PROTOCOL_VERSION } from '#/wire/migration/migration'; +import { AGENT_WIRE_RECORD_KEY, type WireRecord } from '#/wire/record'; import { recordingWireLog, registerTestAgentWire, registerTestEventDispatcher, testWireScope, + noopLogger, } from './stubs'; const SCOPE = 'wire'; @@ -145,3 +152,46 @@ describe('durable observable events', () => { ]); }); }); + +describe('restore from a corrupted journal', () => { + it('folds the valid prefix and heals the journal on disk', async () => { + const ix = disposables.add(new TestInstantiationService()); + ix.set(IEventBus, new SyncDescriptor(EventBusService)); + const bus = ix.get(IEventBus); + const storage = new InMemoryStorageService(); + ix.stub(IFileSystemStorageService, storage); + ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + const log = ix.get(IAppendLogStore); + const scope = testWireScope(SCOPE, KEY); + registerTestAgentWire(ix, scope, { + log, + eventBus: bus, + storage, + logger: noopLogger, + telemetry: noopTelemetryService, + }); + const dispatcher = registerTestEventDispatcher(ix); + const agentState = ix.get(IAgentStateService); + agentState.contributeState(noteKey); + + const metadata = JSON.stringify({ + type: 'metadata', + protocol_version: WIRE_PROTOCOL_VERSION, + created_at: 1, + }); + const note = (text: string): string => + JSON.stringify({ type: 'store-event.note.added', text, time: 2 }); + const healed = `${metadata}\n${note('kept')}\n`; + await storage.write( + scope, + AGENT_WIRE_RECORD_KEY, + new TextEncoder().encode(`${healed}GARBAGE\n${note('dropped')}\n`), + ); + + await dispatcher.restore(); + + expect(agentState.get(noteKey).notes).toEqual(['kept']); + const repaired = await storage.read(scope, AGENT_WIRE_RECORD_KEY); + expect(new TextDecoder().decode(repaired)).toBe(healed); + }); +}); diff --git a/packages/agent-core-v2/test/wire/stubs.ts b/packages/agent-core-v2/test/wire/stubs.ts index 73d9040b6..b081c3b05 100644 --- a/packages/agent-core-v2/test/wire/stubs.ts +++ b/packages/agent-core-v2/test/wire/stubs.ts @@ -1,13 +1,17 @@ import { SyncDescriptor } from '#/_base/di/descriptors'; import { toDisposable } from '#/_base/di/lifecycle'; import type { ServiceRegistration, TestInstantiationService } from '#/_base/di/test'; +import { ILogService } from '#/_base/log/log'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { AgentRuntimeSet } from '#/agent/runtime/agentRuntimeSet'; import { IAgentStateService } from '#/agent/state/agentState'; import { AgentStateService } from '#/agent/state/agentStateService'; import { IAgentScopeContext, makeAgentScopeContext, type IAgentScopeContext as AgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { IEventBus, ISessionEventBus } from '#/app/event/eventBus'; +import { ITelemetryService, noopTelemetryService } from '#/app/telemetry/telemetry'; +import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IEventDispatcher } from '#/state/eventDispatcher'; import { EventDispatcherService } from '#/state/eventDispatcherService'; import { AgentTodo, todoAgentRuntimeProvider } from '#/features/todo/todoAgentRuntime'; @@ -25,6 +29,9 @@ interface TestAgentWireDependencies { readonly log?: IAppendLogStore; readonly blob?: IAgentBlobService; readonly eventBus?: IEventBus; + readonly storage?: IFileSystemStorageService; + readonly logger?: ILogService; + readonly telemetry?: ITelemetryService; } const noopLog: IAppendLogStore = { @@ -51,6 +58,18 @@ const noopEventBus: IEventBus = { subscribe: () => toDisposable(() => {}), }; +export const noopLogger: ILogService = { + _serviceBrand: undefined, + level: 'off', + error: () => {}, + warn: () => {}, + info: () => {}, + debug: () => {}, + child: () => noopLogger, + setLevel: () => {}, + flush: async () => {}, +}; + export function testWireScope(scope: string, journal: string): string { return `${scope}/${journal}`; } @@ -69,6 +88,15 @@ export function registerTestAgentWire( ix.set(IAppendLogStore, dependencies.log ?? noopLog); ix.set(IAgentBlobService, dependencies.blob ?? noopBlob); ix.set(IEventBus, dependencies.eventBus ?? noopEventBus); + if (dependencies.storage !== undefined) { + ix.stub(IFileSystemStorageService, dependencies.storage); + } + if (dependencies.logger !== undefined) { + ix.stub(ILogService, dependencies.logger); + } + if (dependencies.telemetry !== undefined) { + ix.stub(ITelemetryService, dependencies.telemetry); + } ix.set(IWireService, new SyncDescriptor(WireService)); const eventBus = ix.get(IEventBus); if (typeof (eventBus as Partial).activateAgent === 'function') { @@ -85,6 +113,9 @@ export function registerTestAgentWireServices( registration.defineInstance(IAppendLogStore, noopLog); registration.defineInstance(IAgentBlobService, noopBlob); registration.defineInstance(IEventBus, noopEventBus); + registration.defineInstance(IFileSystemStorageService, new InMemoryStorageService()); + registration.defineInstance(ILogService, noopLogger); + registration.defineInstance(ITelemetryService, noopTelemetryService); registration.defineInstance(IAgentStateService, new AgentStateService()); registration.define(IWireService, WireService); registration.define(IEventDispatcher, EventDispatcherService); diff --git a/packages/agent-core-v2/test/wire/wireService.test.ts b/packages/agent-core-v2/test/wire/wireService.test.ts index 8abc654df..012b5a0c4 100644 --- a/packages/agent-core-v2/test/wire/wireService.test.ts +++ b/packages/agent-core-v2/test/wire/wireService.test.ts @@ -4,17 +4,20 @@ import { DisposableStore } from '#/_base/di/lifecycle'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { TestInstantiationService } from '#/_base/di/test'; import { resetUnexpectedErrorHandler, setUnexpectedErrorHandler } from '#/_base/errors/unexpectedError'; +import { ILogService } from '#/_base/log/log'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; +import { ITelemetryService, noopTelemetryService } from '#/app/telemetry/telemetry'; import type { ContentPart } from '#/kosong/contract/message'; import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService, StorageError, StorageErrors } from '#/persistence/interface/storage'; import { WIRE_PROTOCOL_VERSION } from '#/wire/migration/migration'; +import { wireJournalBackupKey } from '#/wire/repair'; import { IWireService } from '#/wire/wire'; import { AGENT_WIRE_RECORD_KEY, type WireRecord } from '#/wire/record'; -import { recordingWireLog, registerTestAgentWire, testWireScope } from './stubs'; +import { recordingWireLog, registerTestAgentWire, testWireScope, noopLogger } from './stubs'; const SCOPE = 'wire'; const KEY = 'journal-test'; @@ -23,14 +26,21 @@ let disposables: DisposableStore; let ix: TestInstantiationService; let wire: IWireService; let log: IAppendLogStore; +let storage: InMemoryStorageService; beforeEach(() => { disposables = new DisposableStore(); ix = disposables.add(new TestInstantiationService()); - ix.stub(IFileSystemStorageService, new InMemoryStorageService()); + storage = new InMemoryStorageService(); + ix.stub(IFileSystemStorageService, storage); ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); log = ix.get(IAppendLogStore); - wire = registerTestAgentWire(ix, testWireScope(SCOPE, KEY), { log }); + wire = registerTestAgentWire(ix, testWireScope(SCOPE, KEY), { + log, + storage, + logger: noopLogger, + telemetry: noopTelemetryService, + }); }); afterEach(() => disposables.dispose()); @@ -376,6 +386,234 @@ describe('WireService readJournal', () => { }); }); +describe('WireService corruption repair', () => { + const enc = new TextEncoder(); + const dec = new TextDecoder(); + const BACKUP_KEY = wireJournalBackupKey(AGENT_WIRE_RECORD_KEY); + + interface RepairCapture { + readonly warnings: Array<{ message: string; payload?: unknown }>; + readonly events: Array<{ name: string; payload: unknown }>; + } + + function wireWithCapture(key: string, capture: RepairCapture): IWireService { + const logger: ILogService = { + _serviceBrand: undefined, + level: 'off', + error: () => {}, + warn: (message, payload) => capture.warnings.push({ message, payload }), + info: () => {}, + debug: () => {}, + child: () => logger, + setLevel: () => {}, + flush: async () => {}, + }; + const telemetry: ITelemetryService = { + ...noopTelemetryService, + track2: (name, payload) => { + capture.events.push({ name, payload }); + }, + }; + const localIx = disposables.add(new TestInstantiationService()); + return registerTestAgentWire(localIx, testWireScope(SCOPE, key), { + log, + storage, + logger, + telemetry, + }); + } + + async function rawBytes(key = AGENT_WIRE_RECORD_KEY): Promise { + const bytes = await storage.read(testWireScope(SCOPE, KEY), key); + return bytes === undefined ? undefined : dec.decode(bytes); + } + + async function seedCorrupt(raw: string): Promise { + await storage.write(testWireScope(SCOPE, KEY), AGENT_WIRE_RECORD_KEY, enc.encode(raw)); + } + + function currentMetadata(createdAt = 1): string { + return JSON.stringify({ + type: 'metadata', + protocol_version: WIRE_PROTOCOL_VERSION, + created_at: createdAt, + }); + } + + it('heals a torn final line, keeps the valid prefix, and backs up the original bytes', async () => { + const valid = `${currentMetadata()}\n${JSON.stringify({ type: 'wire.test.ok', time: 3 })}\n`; + const torn = `${JSON.stringify({ type: 'wire.test.torn', time: 4 }).slice(0, 12)}`; + await seedCorrupt(valid + torn); + + const yielded = await collect(wire.readJournal()); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'wire.test.ok', time: 3 }, + ]); + expect(await rawBytes()).toBe(valid); + expect(await rawBytes(BACKUP_KEY)).toBe(valid + torn); + expect(await collect(wire.readJournal())).toEqual(yielded); + }); + + it('truncates at a corrupted middle line, dropping later valid lines', async () => { + const prefix = `${currentMetadata()}\n${JSON.stringify({ type: 'wire.test.a', time: 1 })}\n`; + const raw = `${prefix}GARBAGE\n${JSON.stringify({ type: 'wire.test.b', time: 2 })}\n`; + await seedCorrupt(raw); + + const yielded = await collect(wire.readJournal()); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'wire.test.a', time: 1 }, + ]); + expect(await rawBytes()).toBe(prefix); + expect(await rawBytes(BACKUP_KEY)).toBe(raw); + }); + + it('does not surface AppendLogCorruptedError from the restore read path', async () => { + await seedCorrupt(`${currentMetadata()}\nGARBAGE\n`); + + await expect(collect(wire.readJournal())).resolves.toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + ]); + }); + + it('keeps the first backup when the journal corrupts again after a repair', async () => { + const first = `${currentMetadata()}\nGARBAGE-1\n`; + await seedCorrupt(first); + await collect(wire.readJournal()); + const second = `${currentMetadata()}\nGARBAGE-2\n`; + await seedCorrupt(second); + + await collect(wire.readJournal()); + + expect(await rawBytes(BACKUP_KEY)).toBe(first); + }); + + it('repairs through the migration rewrite path when corruption meets an old version', async () => { + const legacy = `${JSON.stringify({ type: 'metadata', protocol_version: '1.4', created_at: 1 })}\n`; + const raw = `${legacy}${JSON.stringify({ type: 'wire.test.legacy', time: 9 })}\nGARBAGE\n`; + await seedCorrupt(raw); + + const yielded = await collect(wire.readJournal()); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + { type: 'wire.test.legacy', time: 9 }, + ]); + expect(await rawBytes()).toBe( + `${currentMetadata()}\n${JSON.stringify({ type: 'wire.test.legacy', time: 9 })}\n`, + ); + expect(await rawBytes(BACKUP_KEY)).toBe(raw); + }); + + it('reports a corrupted middle line through a warn log and the wire_repair event', async () => { + const capture: RepairCapture = { warnings: [], events: [] }; + const svc = wireWithCapture(KEY, capture); + const prefix = `${currentMetadata()}\n${JSON.stringify({ type: 'wire.test.a', time: 1 })}\n`; + const raw = `${prefix}GARBAGE\n${JSON.stringify({ type: 'wire.test.b', time: 2 })}\n`; + await seedCorrupt(raw); + + await collect(svc.readJournal()); + + expect(capture.warnings).toHaveLength(1); + expect(capture.warnings[0]!.message).toBe( + 'corrupted wire journal truncated to its valid prefix', + ); + expect(capture.warnings[0]!.payload).toMatchObject({ + lineNumber: 3, + reason: 'corrupted', + outcome: 'repaired', + droppedCount: 2, + backupCreated: true, + }); + expect(capture.events).toEqual([ + { + name: 'wire_repair', + payload: { + kind: 'corrupted', + outcome: 'repaired', + dropped_count: 2, + backup_created: true, + }, + }, + ]); + }); + + it('counts dropped lines from the damaged line when the repair inserts metadata', async () => { + const capture: RepairCapture = { warnings: [], events: [] }; + const svc = wireWithCapture(KEY, capture); + const raw = `${JSON.stringify({ type: 'wire.test.legacy', time: 9 })}\nGARBAGE\n`; + await seedCorrupt(raw); + + await collect(svc.readJournal()); + + expect(capture.events).toEqual([ + { + name: 'wire_repair', + payload: { + kind: 'corrupted', + outcome: 'repaired', + dropped_count: 1, + backup_created: true, + }, + }, + ]); + }); + + it('reports a torn tail as truncation through the wire_repair event', async () => { + const capture: RepairCapture = { warnings: [], events: [] }; + const svc = wireWithCapture(KEY, capture); + const raw = `${currentMetadata()}\n${JSON.stringify({ type: 'wire.test.a' }).slice(0, 10)}`; + await seedCorrupt(raw); + + await collect(svc.readJournal()); + + expect(capture.events).toEqual([ + { + name: 'wire_repair', + payload: { + kind: 'truncated', + outcome: 'repaired', + dropped_count: 1, + backup_created: true, + }, + }, + ]); + }); + + it('keeps restoring from the valid prefix when the on-disk repair itself fails', async () => { + const capture: RepairCapture = { warnings: [], events: [] }; + const svc = wireWithCapture(KEY, capture); + const prefix = `${currentMetadata()}\n`; + const raw = `${prefix}GARBAGE\n`; + await seedCorrupt(raw); + const originalWrite = storage.write.bind(storage); + storage.write = async (scope, key, data, options) => { + if (key === AGENT_WIRE_RECORD_KEY) throw new Error('disk full'); + return originalWrite(scope, key, data, options); + }; + + const yielded = await collect(svc.readJournal()); + + expect(yielded).toEqual([ + { type: 'metadata', protocol_version: WIRE_PROTOCOL_VERSION, created_at: 1 }, + ]); + expect(capture.events).toEqual([ + { + name: 'wire_repair', + payload: { + kind: 'corrupted', + outcome: 'failed', + dropped_count: 1, + backup_created: true, + }, + }, + ]); + }); +}); + describe('WireService flush', () => { it('drains the dehydrate queue before resolving', async () => { const records: WireRecord[] = []; diff --git a/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts b/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts index 9145b8fbd..030e7a6a8 100644 --- a/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts +++ b/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts @@ -58,7 +58,7 @@ class TestRuntimeUnitHost implements RuntimeUnitHost { try { attachment = await prepare(host); } catch (error) { - for (const registration of registrations.reverse()) await registration.remove(); + for (const registration of registrations.toReversed()) await registration.remove(); throw error; } let active = true; @@ -66,7 +66,7 @@ class TestRuntimeUnitHost implements RuntimeUnitHost { if (!active) return; active = false; await attachment.dispose(); - for (const registration of registrations.reverse()) await registration.remove(); + for (const registration of registrations.toReversed()) await registration.remove(); const index = this.units.indexOf(handle); if (index >= 0) this.units.splice(index, 1); }; @@ -88,7 +88,7 @@ class TestRuntimeUnitHost implements RuntimeUnitHost { } async dispose(): Promise { - for (const unit of [...this.units].reverse()) await unit.dispose(); + for (const unit of [...this.units].toReversed()) await unit.dispose(); } } @@ -139,7 +139,7 @@ function manager( { scope: () => 'sessions' }, workspaces, { ready }, - ...Array.from({ length: 22 }, () => undefined), + ...Array.from({ length: 23 }, () => undefined), new TestRuntimeUnitHostFactory(), ]; args[20] = { entries: () => [] }; @@ -198,7 +198,7 @@ describe('WorkspaceInstanceManager', () => { await remote.dispose(); expect(one.runtimes.current('remote')).toBeUndefined(); expect(two.runtimes.current('remote')).toBeUndefined(); - expect(events.filter((event) => event.startsWith('detach:remote-provider:')).sort()).toEqual([ + expect(events.filter((event) => event.startsWith('detach:remote-provider:')).toSorted()).toEqual([ 'detach:remote-provider:one', 'detach:remote-provider:two', ]); diff --git a/packages/agent-gateway/test/sessions.test.ts b/packages/agent-gateway/test/sessions.test.ts index 3ca73f9b5..a0c2e5d5b 100644 --- a/packages/agent-gateway/test/sessions.test.ts +++ b/packages/agent-gateway/test/sessions.test.ts @@ -1,5 +1,5 @@ import { randomBytes } from 'node:crypto'; -import { mkdir, mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'; +import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { inflateRawSync } from 'node:zlib'; @@ -830,6 +830,48 @@ describe('server-v2 /api/v1/sessions', () => { ]); }); + it('fork heals a corrupted source wire through the shared repair path', async () => { + const cwd = home as string; + const parent = await postJson('/api/v1/sessions', { metadata: { cwd } }); + const parentId = parent.body.data.id; + const session = getLiveSessionById((server as RunningServer).core.accessor, parentId); + expect(session).toBeDefined(); + const mainContext = await session!.accessor.get(IAgentLifecycleService).create({ agentId: MAIN_AGENT_ID }); + const cron = session!.accessor.get(IAgentLifecycleService).resolve(mainContext, AgentCron); + const task = cron.addTask({ cron: '0 9 * * *', prompt: 'survives corruption', recurring: true }); + await closeSessionById((server as RunningServer).core.accessor, parentId); + + const wireRelatives = (await readdir(home as string, { recursive: true })).filter((path) => + path.endsWith(join(parentId, 'agents', 'main', 'wire.jsonl')), + ); + expect(wireRelatives).toHaveLength(1); + const wirePath = join(home as string, wireRelatives[0]!); + const originalLines = (await readFile(wirePath, 'utf8')) + .split('\n') + .filter((line) => line.length > 0); + expect(originalLines.length).toBeGreaterThan(1); + const corrupted = `${[...originalLines, 'GARBAGE'].join('\n')}\n`; + await writeFile(wirePath, corrupted); + + const forked = await postJson(`/api/v1/sessions/${parentId}:fork`, {}); + expect(forked.body.code).toBe(0); + + expect(await readFile(wirePath, 'utf8')).toBe(`${originalLines.join('\n')}\n`); + expect(await readFile(`${wirePath}.bak`, 'utf8')).toBe(corrupted); + + const forkedId = forked.body.data.id; + const forkedLive = getLiveSessionById((server as RunningServer).core.accessor, forkedId); + expect(forkedLive).toBeDefined(); + const forkedManager = forkedLive!.accessor.get(IAgentLifecycleService); + const forkedCron = forkedManager.resolve(forkedManager.get(MAIN_AGENT_ID)!, AgentCron); + expect(forkedCron.list().map((t) => ({ id: t.id, prompt: t.prompt }))).toEqual([ + { id: task.id, prompt: 'survives corruption' }, + ]); + + const fetched = await getJson(`/api/v1/sessions/${forkedId}`); + expect(fetched.body.code).toBe(0); + }); + it('keeps cron tasks across a server restart through the wire', async () => { const cwd = home as string; const parent = await postJson('/api/v1/sessions', { metadata: { cwd } });