Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
21 changes: 9 additions & 12 deletions hosts/kimi/transcript.js
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,16 @@ const validResult = (result) => result && typeof result === 'object' && typeof r
const pathOf = (input, cwd) => text(input?.path)
&& (isAbsolute(input.path) ? input.path : text(cwd) && isAbsolute(cwd) ? resolve(cwd, input.path) : null);

// The newest turns' tasks are the ones a turn still waiting for its end can name; older ones are past their end.
const MAX_TASKS = 64;
const backgroundTaskOf = (result) => taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1]);

/** Only discovers fixed task IDs; the reader chooses a bounded file under the known session root. */
export function taskIdsOf(records = []) {
const ids = [];
for (const record of records.filter(main)) {
const event = record.type === 'context.append_loop_event' ? record.event : null;
if (event?.type === 'tool.call' && event.name === 'TaskOutput') ids.push(taskId(event.args?.task_id));
if (event?.type === 'tool.result') {
ids.push(taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(event.result?.output))?.[1]));
}
}
/**
* Only discovers fixed task IDs, those the requested turn's Bash calls started: no other task can settle its
* evidence. The reader chooses a bounded file under the known session root.
*/
export function taskIdsOf(records = [], { turnId } = {}) {
const ids = callsOf(records, nativeTurnIdOf(turnId)).filter((call) => call.name === 'Bash')
.map((call) => backgroundTaskOf(resultOf(records, call)));
const newestLast = [...new Set(ids.filter(Boolean).reverse())].reverse();
return newestLast.slice(-MAX_TASKS);
}
Expand Down Expand Up @@ -63,7 +60,7 @@ function notesOf(records, tasks, turnId, cwd) {
const result = resultOf(records, call);
const input = call.args && typeof call.args === 'object' ? call.args : {};
if (call.name === 'Bash' && text(input.command)) {
const expectedTask = taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1]);
const expectedTask = backgroundTaskOf(result);
const matching = tasks.filter((task) => expectedTask && task?.kind === 'process'
&& task.parentToolCallId === call.toolCallId && task.command === input.command && task.taskId === expectedTask);
const task = matching.length === 1 ? matching[0] : null;
Expand Down
6 changes: 3 additions & 3 deletions lib/runtime/session-log.js
Original file line number Diff line number Diff line change
Expand Up @@ -132,8 +132,8 @@ function sessionDirOf(declaration, home, nativeId) {
* names the log home (`root(env)`), its workspace index (`workspaceIndex`, listed as `[id, projectDir]` by
* `workspacesOf`), a session's folder in a workspace (`sessionDir(workspaceId, nativeId)`) and its log (`logFile`);
* `nativeSessionOf(sessionId, scope)` takes the plugin's session id back to the host's. `decode(records, { turnId,
* tasks, projectDir, unread })` answers the row, where `unread` is the byte span of a long log left unread; `taskIds(records)` and `tasksDir(sessionDir)` name the background task files
* it also reads. `scope` is a pseudonym of the log home, for session ids, so no path reaches a Hub task id. Anything
* tasks, projectDir, unread })` answers the row, where `unread` is the byte span of a long log left unread;
* `taskIds(records, { turnId })` and `tasksDir(sessionDir)` name the background task files it also reads. `scope` is a pseudonym of the log home, for session ids, so no path reaches a Hub task id. Anything
* unreadable answers null, never an empty valid turn.
*/
export function sessionLogReaderOf(declaration, env = process.env) {
Expand All @@ -160,7 +160,7 @@ export function sessionLogReaderOf(declaration, env = process.env) {
const key = JSON.stringify([session.dir, stat.dev, stat.ino, stat.size, stat.mtimeMs]);
if (parsed?.key !== key) parsed = { key, ...recordsOf(fd, stat.size) };
const { records, unread } = parsed;
const tasks = tasksOf(root, declaration.tasksDir?.(session.dir), declaration.taskIds?.(records) ?? []);
const tasks = tasksOf(root, declaration.tasksDir?.(session.dir), declaration.taskIds?.(records, { turnId }) ?? []);
if (!tasks) return null;
return declaration.decode(records, { turnId, tasks, projectDir: session.projectDir, unread }) ?? null;
} catch {
Expand Down
23 changes: 17 additions & 6 deletions test/hosts/kimi/readers.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -91,13 +91,24 @@ test('unsupported schemas and foreign agents do not supply evidence', () => {
assert.equal(status([meta, { ...prompt(0), origin: { kind: 'cron' } }]), null);
});

test('fixed task discovery rejects paths and does not inspect arbitrary output paths', () => {
const records = [event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: 'bash-task' } }), event({ type: 'tool.result', result: { output: 'task_id: bash-task\nstatus: running' } }), event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: '../outside' } })];
assert.deepEqual(taskIdsOf(records), ['bash-task']);
const many = Array.from({ length: 70 }, (_, index) => event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: `task-${index}` } }));
const ids = taskIdsOf([...many, many[0]]);
test('fixed task discovery reads only background tasks the requested turn started and rejects paths', () => {
const bash = (id, task, turnId = '0') => [
event({ type: 'tool.call', uuid: `call-${id}`, toolCallId: `tool-${id}`, turnId, name: 'Bash', args: { command: 'npm test' } }),
event({ type: 'tool.result', parentUuid: `call-${id}`, toolCallId: `tool-${id}`, result: { output: `task_id: ${task}\nstatus: running`, durationMs: 1 } }),
];
const records = [
...bash(0, 'bash-task'),
...bash(1, '../outside'),
...bash(2, 'foreign-task', '1'),
event({ type: 'tool.call', uuid: 'read-uuid', toolCallId: 'read-call', turnId: '0', name: 'TaskOutput', args: { task_id: 'read-task' } }),
event({ type: 'tool.result', result: { output: 'task_id: unrelated-task\nstatus: running' } }),
];
assert.deepEqual(taskIdsOf(records, { turnId: '0' }), ['bash-task']);
assert.deepEqual(taskIdsOf(records, { turnId: '1' }), ['foreign-task']);
const many = Array.from({ length: 70 }, (_, index) => bash(index, `task-${index}`)).flat();
const ids = taskIdsOf([...many, ...bash(70, 'task-0')], { turnId: '0' });
assert.equal(ids.length, 64);
assert.equal(ids.at(-1), 'task-0', 'a task the newest turn names again is kept');
assert.equal(ids.at(-1), 'task-0', 'a task named again by a later call is kept');
});

test('duplicate native events do not duplicate checks and mismatched results remain unknown', () => {
Expand Down
33 changes: 33 additions & 0 deletions test/runtime/session-log.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';

import { taskIdsOf, turnStatusOf } from '../../hosts/kimi/transcript.js';

import { recordReaderOf } from '../../lib/runtime/records.js';

function fixture(t, overrides = {}) {
Expand Down Expand Up @@ -137,3 +139,34 @@ test('a task the log names many times counts once toward the task limit', (t) =>
assert.ok(f.reader(f.sessionId, '0'));
assert.deepEqual(f.decoded.at(-1).context.tasks, [{ taskId: 'task', status: 'running' }]);
});

test('a damaged task of an earlier turn does not hide a later completed turn, even one that reads its output', (t) => {
const f = fixture(t, {
taskIds: taskIdsOf, tasksDir: (dir) => join(dir, 'agents', 'main', 'tasks'), decode: turnStatusOf,
});
const prompt = (turnId) => ({ type: 'turn.prompt', agentId: 'main', turnId, promptId: `prompt-${turnId}`, origin: { kind: 'user' } });
const event = (value) => ({ type: 'context.append_loop_event', agentId: 'main', event: value });
const records = [
{ type: 'metadata', protocol_version: '1.5' },
prompt(0),
event({ type: 'tool.call', uuid: 'old-call', turnId: '0', toolCallId: 'old-check', name: 'Bash', args: { command: 'npm test', run_in_background: true } }),
event({ type: 'tool.result', parentUuid: 'old-call', toolCallId: 'old-check', result: { output: 'task_id: old-task\nstatus: running', durationMs: 1 } }),
{ type: 'turn.ended', agentId: 'main', turnId: 0, reason: 'completed' },
prompt(1),
event({ type: 'tool.call', uuid: 'output-call', turnId: '1', toolCallId: 'old-output', name: 'TaskOutput', args: { task_id: 'old-task' } }),
event({ type: 'tool.result', parentUuid: 'output-call', toolCallId: 'old-output', result: { output: 'status: completed', durationMs: 1 } }),
event({ type: 'tool.call', uuid: 'current-call', turnId: '1', toolCallId: 'current-check', name: 'Bash', args: { command: 'npm test' } }),
event({ type: 'tool.result', parentUuid: 'current-call', toolCallId: 'current-check', result: { output: '# pass 1\n# fail 0', durationMs: 1 } }),
{ type: 'turn.ended', agentId: 'main', turnId: 1, reason: 'completed' },
];
writeFileSync(f.wire, records.map(line).join(''));
mkdirSync(join(f.agent, 'tasks'));
const taskFile = join(f.agent, 'tasks', 'old-task.json');
writeFileSync(taskFile, JSON.stringify({ taskId: 'old-task', kind: 'process', parentToolCallId: 'old-check', command: 'npm test', status: 'completed', exitCode: 0 }));
assert.equal(f.reader(f.sessionId, '1').status, 'completed');
writeFileSync(taskFile, '{');
assert.equal(f.reader(f.sessionId, '0'), null, 'the requested turn still requires its own task to be readable');
const current = f.reader(f.sessionId, '1');
assert.equal(current?.status, 'completed');
assert.deepEqual(current.record.checks, [{ command: 'npm test', exit: null, failed: false }]);
});
Loading