Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/print-wire-flush-on-exit.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@moonshot-ai/kimi-code": patch
---

Fix print mode (`kimi -p`) losing a failed turn's closing session records on error exit.
30 changes: 30 additions & 0 deletions apps/kimi-code/src/cli/v2/run-v2-print.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import {
IBootstrapService,
IConfigService,
IEventBus,
IEventDispatcher,
IOAuthToolkit,
ISessionIndex,
ISessionManager,
Expand Down Expand Up @@ -196,6 +197,7 @@ export async function runV2Print(
}

let restorePermission = async (): Promise<void> => {};
let flushWires = async (): Promise<void> => {};
let removeTerminationCleanup: (() => void) | undefined;
let cleanupPromise: Promise<void> | undefined;
let telemetryService: ITelemetryService | undefined;
Expand All @@ -205,6 +207,13 @@ export async function runV2Print(
setCrashPhase('shutdown');
try {
await restorePermission();
// The turn's tail records (step.end / turn.ended / prompt.completed) are
// dispatched fire-and-forget and reach the journal only through the wire
// service's async persist queue. Without an explicit flush, a cleanup
// that returns fast (e.g. telemetry disabled) lets process.exit cut off
// that queue before the records land on disk. Best-effort: a persist
// failure was already reported where the append failed.
await raceWithTimeout(flushWires(), CLI_SHUTDOWN_TIMEOUT_MS).catch(() => {});

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep sequential shutdown within the cleanup deadline

When the wire flush uses its 3-second allowance and both telemetry pipelines are also slow, cleanup can now take up to 9 seconds (flushWires, telemetryService.shutdown, and shutdownTelemetry each receive CLI_SHUTDOWN_TIMEOUT_MS) while the outer cleanup still stops waiting after 8 seconds. On a failed run or signal, the caller then invokes process.exit immediately, truncating the final telemetry shutdown and potentially preventing app.dispose() from running. Use a shared deadline, run independent shutdowns concurrently, or increase the outer bound to cover all sequential phases.

Useful? React with 👍 / 👎.

} finally {
if (telemetryService !== undefined) {
await raceWithTimeout(telemetryService.shutdown(), CLI_SHUTDOWN_TIMEOUT_MS);
Expand Down Expand Up @@ -254,6 +263,7 @@ export async function runV2Print(

const resolved = await resolveNativeSession(app, opts, workDir, defaultModel, stderr);
restorePermission = resolved.restorePermission;
flushWires = () => flushSessionWires(resolved.session, resolved.agent);

telemetryService.setContext({ session_id: resolved.session.id, model: resolved.telemetryModel });
setTelemetryContext({ sessionId: resolved.session.id });
Expand Down Expand Up @@ -869,6 +879,26 @@ function countPendingBackgroundTasks(session: ISessionScopeHandle): number {
return count;
}

/**
* Flush every session agent's wire journal. The main agent handle is included
* explicitly: `IAgentLifecycleService` skips `closing` agents, but a closing
* agent's already-dispatched tail records still deserve to land on disk.
*/
async function flushSessionWires(
session: ISessionScopeHandle,
mainAgent: IAgentScopeHandle,
): Promise<void> {
const agentManager = session.accessor.get(IAgentLifecycleService);
const handles = new Set<IAgentScopeHandle>([mainAgent]);
for (const agent of agentManager.list()) {
const handle = agentManager.handleOf(agent.agentId);
if (handle !== undefined) handles.add(handle);
Comment on lines +903 to +905

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Include closing subagents in the wire flush

When a non-main agent has entered AgentLifecycleService.remove() during shutdown, both list() and handleOf() exclude it because its managed.closing flag is already set, so this collector omits that agent from both quiescence and flushSessionWires(). Its removal releases the append-log lease with a fire-and-forget retirement flush, allowing the signal handler's subsequent process.exit() to truncate that subagent's closing records. Fresh evidence beyond the earlier all-agent comments is the explicit closing-agent filter in both lifecycle lookup methods; only the main agent is preserved separately here.

Useful? React with 👍 / 👎.

}
await Promise.all(
[...handles].map((handle) => handle.accessor.get(IEventDispatcher).flush()),
);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep flushing agents after one journal fails

When multiple agents are present and any dispatcher's flush() rejects before another agent finishes flushing, Promise.all rejects immediately; the caller swallows that rejection and proceeds to app.dispose(), so the remaining flushes are no longer awaited and the subsequent process exit can still truncate otherwise healthy journals. Settle every per-agent flush independently (for example with Promise.allSettled) before returning the best-effort cleanup.

Useful? React with 👍 / 👎.

}

async function drainBackgroundTasks(
session: ISessionScopeHandle,
ceilingS: number | undefined,
Expand Down
87 changes: 87 additions & 0 deletions apps/kimi-code/test/cli/v2-run-print.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import {
IBootstrapService,
IConfigService,
IEventBus,
IEventDispatcher,
IFileSystemStorageService,
IOAuthToolkit,
ISessionIndex,
Expand Down Expand Up @@ -186,6 +187,7 @@ function makeFakeHarness() {
[IAgentTaskService, { list: vi.fn(() => []) }],
[IAgentCronService, { getNextFireTime: vi.fn(() => null) }],
[IAgentGoalService, goal],
[IEventDispatcher, { flush: vi.fn(async () => {}) }],
[
IAgentScopeContext,
makeAgentScopeContext({ agentId: 'main', agentScope: 'agents/main' }),
Expand Down Expand Up @@ -613,4 +615,89 @@ describe('runV2Print', () => {
expect(initOrder).toBeDefined();
expect(reconcileOrder).toBeGreaterThan(initOrder!);
});

it('flushes the wire journal before disposing the app', async () => {
const stdout = writer();
const stderr = writer();
const { app, agentServices } = makeFakeHarness();

mocks.bootstrap.mockReturnValue({ app });
mocks.ensureMainAgent.mockResolvedValue({ agentId: 'main', generation: 1 });

await runV2Print(opts() as never, '1.2.3-test', { stdout, stderr });

const dispatcher = agentServices.get(IEventDispatcher) as {
flush: ReturnType<typeof vi.fn>;
};
expect(dispatcher.flush).toHaveBeenCalled();
const flushOrder = dispatcher.flush.mock.invocationCallOrder[0];
const disposeOrder = app.dispose.mock.invocationCallOrder[0];
expect(flushOrder).toBeDefined();
expect(disposeOrder).toBeGreaterThan(flushOrder!);
});

it('flushes the wire journal when the turn fails', async () => {
const stdout = writer();
const stderr = writer();
const { app, agentServices } = makeFakeHarness();

const promptService = agentServices.get(IAgentPromptService) as {
enqueue: ReturnType<typeof vi.fn>;
};
promptService.enqueue.mockResolvedValueOnce({
launched: Promise.resolve({
id: 1,
result: Promise.resolve({
type: 'failed',
error: { code: 'provider.overloaded', message: 'llm request failed' },
}),
}),
});

mocks.bootstrap.mockReturnValue({ app });
mocks.ensureMainAgent.mockResolvedValue({ agentId: 'main', generation: 1 });

await expect(runV2Print(opts() as never, '1.2.3-test', { stdout, stderr })).rejects.toThrow(
'provider.overloaded: llm request failed',
);

const dispatcher = agentServices.get(IEventDispatcher) as {
flush: ReturnType<typeof vi.fn>;
};
expect(dispatcher.flush).toHaveBeenCalled();
const flushOrder = dispatcher.flush.mock.invocationCallOrder[0];
const disposeOrder = app.dispose.mock.invocationCallOrder[0];
expect(disposeOrder).toBeGreaterThan(flushOrder!);
});

it('does not let a wire flush failure mask the turn outcome', async () => {
const stdout = writer();
const stderr = writer();
const { app, agentServices } = makeFakeHarness();

const promptService = agentServices.get(IAgentPromptService) as {
enqueue: ReturnType<typeof vi.fn>;
};
promptService.enqueue.mockResolvedValueOnce({
launched: Promise.resolve({
id: 1,
result: Promise.resolve({
type: 'failed',
error: { code: 'provider.overloaded', message: 'llm request failed' },
}),
}),
});
const dispatcher = agentServices.get(IEventDispatcher) as {
flush: ReturnType<typeof vi.fn>;
};
dispatcher.flush.mockRejectedValueOnce(new Error('disk full'));

mocks.bootstrap.mockReturnValue({ app });
mocks.ensureMainAgent.mockResolvedValue({ agentId: 'main', generation: 1 });

await expect(runV2Print(opts() as never, '1.2.3-test', { stdout, stderr })).rejects.toThrow(
'provider.overloaded: llm request failed',
);
expect(app.dispose).toHaveBeenCalled();
});
});
Loading