From 7426296899bff27aa334e4a48f89e32292751a7e Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:05:21 +0900 Subject: [PATCH 01/12] docs: record Pi finality repair roadmap --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 9cde51f49..80b8290c9 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 9cde51f4943687a9495767ad71d70447ef1a3fe2 +Subproject commit 80b8290c960eb1f801f231a32c84f658041b15ae From 360e2a847a2ec4921b6e98a27c9f2f272a0c58c9 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:09:53 +0900 Subject: [PATCH 02/12] fix(pi): select typed final response at session settlement --- src/agent/pi-runtime.ts | 92 +++++++---- src/agent/runtime/pi-turn.ts | 135 ++++++++++++++++ structure/str_func.md | 2 +- tests/unit/pi-finality-runtime.test.ts | 206 +++++++++++++++++++++++++ 4 files changed, 406 insertions(+), 29 deletions(-) create mode 100644 src/agent/runtime/pi-turn.ts create mode 100644 tests/unit/pi-finality-runtime.test.ts diff --git a/src/agent/pi-runtime.ts b/src/agent/pi-runtime.ts index d19700a96..322b55705 100644 --- a/src/agent/pi-runtime.ts +++ b/src/agent/pi-runtime.ts @@ -11,6 +11,8 @@ import { probeOpenCodexEndpointModels } from '../cli/opencodex-models.js'; import { launchSpec } from '../core/exec-name.js'; import { mergeEnvWindowsSafe } from './spawn-env.js'; import { createTextStreamReader } from './stream-text.js'; +import type { RuntimeTurnOutcome } from '../shared/runtime-contract.js'; +import { PiTurnAccumulator, PiRuntimeError, piSupportsSettled } from './runtime/pi-turn.js'; export type PiProfileMode = 'basic' | 'openai' | 'anthropic' | 'vertex'; export type PiApiKind = 'openai-completions' | 'openai-responses' | 'anthropic-messages' | 'google-vertex'; @@ -60,12 +62,18 @@ export interface PiRpcSession { effort?: string; onEvent?: (event: PiRuntimeEvent) => void; onRawRecord?: (record: unknown) => void; - }): Promise<{ text: string; stderr: string }>; + }): Promise; abort(): Promise; close(): void; kill(): void; } +export interface PiPromptResult { + text: string; + stderr: string; + runtimeOutcome?: RuntimeTurnOutcome; +} + function notifyPiRawRecord(observer: ((record: unknown) => void) | undefined, record: unknown): void { try { observer?.(record); } catch { console.warn('[jaw:pi] raw activity observer failed'); } @@ -286,6 +294,11 @@ function resolvePiCommandIdentity(command: PiCommand, env: NodeJS.ProcessEnv = p return JSON.stringify({ source: command.source, command: command.command, baseArgs: command.baseArgs, version }); } +function usesPiSettled(command: PiCommand): boolean { + const identity = JSON.parse(resolvePiCommandIdentity(command)) as { version: string }; + return piSupportsSettled(identity.version); +} + function loadPiAbortEffective(profileId: string, command: PiCommand): boolean { try { const raw = JSON.parse(fs.readFileSync(join(JAW_HOME, 'pi', 'rpc-capabilities.json'), 'utf8')) as unknown; @@ -493,11 +506,12 @@ function extractPiSessionId(obj: Record): string { } type PersistentPrompt = { - text: string; + turn: PiTurnAccumulator; + requestId: number; stderrStart: number; onEvent: ((event: PiRuntimeEvent) => void) | undefined; onRawRecord: ((record: unknown) => void) | undefined; - resolve: (result: { text: string; stderr: string }) => void; + resolve: (result: PiPromptResult) => void; reject: (error: Error) => void; }; @@ -527,6 +541,7 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options }): PiRpcSession { const dir = ensurePiRuntimeConfig(pi, profile.id, options.effort || '', options.root); const cmd = resolvePiCommand(); + const settledProtocol = usesPiSettled(cmd); const args = [ ...cmd.baseArgs, '--mode', 'rpc', @@ -608,15 +623,23 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options } if (event.tool) activePrompt?.onEvent?.({ kind: 'tool', ...event.tool }); if (event.thinking) activePrompt?.onEvent?.({ kind: 'thinking', text: event.thinking }); - if (event.text && activePrompt && !(event.done && activePrompt.text)) { - activePrompt.text += event.text; - activePrompt.onEvent?.({ kind: 'text', text: event.text }); + if (activePrompt && record['type'] === 'response' && record['command'] === 'prompt' + && record['id'] === activePrompt.requestId && record['success'] === false) { + const message = typeof record['error'] === 'string' ? record['error'] + : trimString((record['error'] as Record | undefined)?.['message']); + rejectOutstanding(new Error(message || 'pi rpc prompt rejected')); + return; } - if (event.done || (abortWait && nonRunning)) { + const accepted = activePrompt?.turn.observe(record); + if (accepted?.text) activePrompt?.onEvent?.({ kind: 'text', text: accepted.text }); + const abortNonRunning = Boolean(isAbortResponse) && nonRunning && !abortWait?.terminal; + const terminal = accepted?.done || (!activePrompt && record['type'] === (settledProtocol ? 'agent_settled' : 'agent_end')); + if (terminal || abortNonRunning) { if (activePrompt) { const prompt = activePrompt; activePrompt = null; - prompt.resolve({ text: prompt.text, stderr: stderr.slice(prompt.stderrStart) }); + const runtimeOutcome = prompt.turn.snapshot(abortNonRunning ? 'stopped' : undefined); + prompt.resolve({ text: runtimeOutcome.partialText, stderr: stderr.slice(prompt.stderrStart), runtimeOutcome }); } if (abortWait) abortWait.terminal = true; } @@ -626,7 +649,7 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options if (activePrompt) { const prompt = activePrompt; activePrompt = null; - prompt.reject(error); + prompt.reject(new PiRuntimeError(error, prompt.turn.snapshot(child.killed ? 'stopped' : 'error'))); } if (abortWait) { const wait = abortWait; @@ -651,13 +674,13 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options // A head-only cap without this reset would freeze stderr.length // and make every later slice() return ''. stderr = ''; - activePrompt = { text: '', stderrStart: stderr.length, onEvent: opts.onEvent, onRawRecord: opts.onRawRecord, resolve, reject }; + activePrompt = { turn: new PiTurnAccumulator(settledProtocol), requestId: 0, stderrStart: stderr.length, + onEvent: opts.onEvent, onRawRecord: opts.onRawRecord, resolve, reject }; try { if (opts.effort) write('set_thinking_level', { level: opts.effort }); - write('prompt', { message }); + activePrompt.requestId = write('prompt', { message }); } catch (error) { - activePrompt = null; - reject(error as Error); + rejectOutstanding(error as Error); } }); }, @@ -728,9 +751,10 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: { onEvent?: (event: PiRuntimeEvent) => void; onRawRecord?: (record: unknown) => void; root?: string; -}): { child: ChildProcess; done: Promise<{ text: string; code: number; sessionId?: string | null; stderr: string }> } { +}): { child: ChildProcess; done: Promise } { const dir = ensurePiRuntimeConfig(pi, profile.id, options.effort || '', options.root); const cmd = resolvePiCommand(); + const turn = new PiTurnAccumulator(usesPiSettled(cmd)); const args = [ ...cmd.baseArgs, '--mode', 'rpc', @@ -751,23 +775,25 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: { const decoder = new StringDecoder('utf8'); let buffer = ''; let stderr = ''; - let text = ''; const stderrReader = createTextStreamReader(); let sessionId: string | null = null; let doneSettled = false; let seq = 1; - const done = new Promise<{ text: string; code: number; sessionId?: string | null; stderr: string }>((resolve, reject) => { - const finish = (code = 0) => { + let promptId = 0; + const done = new Promise((resolve, reject) => { + const finish = (code = 0, status?: RuntimeTurnOutcome['status']) => { if (doneSettled) return; doneSettled = true; try { child.stdin.end(); } catch { /* already closed */ } setTimeout(() => { if (!child.killed && child.exitCode == null) child.kill('SIGTERM'); }, 750); - resolve({ text, code, sessionId, stderr }); + const runtimeOutcome = turn.snapshot(status); + resolve({ text: runtimeOutcome.partialText, code, sessionId, stderr, runtimeOutcome }); }; let parseFailures = 0; const dispatchLine = (line: string) => { + if (doneSettled) return; let parsed: unknown; try { parsed = JSON.parse(line); } catch { @@ -776,6 +802,12 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: { return; } notifyPiRawRecord(options.onRawRecord, parsed); + const record = parsed && typeof parsed === 'object' ? parsed as Record : {}; + if (record['type'] === 'response' && record['command'] === 'prompt' + && record['id'] === promptId && record['success'] === false) { + finish(1, 'error'); + return; + } const event = parsePiRpcRecord(parsed); if (event.sessionId) { sessionId = event.sessionId; @@ -783,13 +815,15 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: { } if (event.tool) options.onEvent?.({ kind: 'tool', ...event.tool }); if (event.thinking) options.onEvent?.({ kind: 'thinking', text: event.thinking }); - if (event.text && !(event.done && text)) { - text += event.text; - options.onEvent?.({ kind: 'text', text: event.text }); - } - if (event.done) finish(0); + const accepted = turn.observe(parsed); + if (accepted.text) options.onEvent?.({ kind: 'text', text: accepted.text }); + if (accepted.done) finish(0); }; - child.on('error', reject); + child.on('error', error => { + if (doneSettled) return; + doneSettled = true; + reject(new PiRuntimeError(error, turn.snapshot('error'))); + }); child.stdout.on('data', (chunk) => { buffer += decoder.write(chunk); const lines = buffer.split('\n'); @@ -802,19 +836,21 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: { for (const line of lines) if (line.trim()) dispatchLine(line.trim()); }); child.stderr.on('data', (chunk) => { if (stderr.length < 4000) stderr += stderrReader.write(chunk); }); - child.on('close', (code) => { + child.on('close', (code, signal) => { if (buffer.trim()) dispatchLine(buffer.trim()); - finish(code ?? 0); + finish(code ?? 1, signal || child.killed ? 'stopped' : 'error'); }); }); const write = (type: string, fields: Record = {}) => { - child.stdin.write(JSON.stringify({ id: seq++, type, ...fields }) + '\n'); + const id = seq++; + child.stdin.write(JSON.stringify({ id, type, ...fields }) + '\n'); + return id; }; write('get_state'); if (options.effort) write('set_thinking_level', { level: options.effort }); const fullPrompt = options.sysPrompt ? `${options.sysPrompt}\n\n${options.prompt}` : options.prompt; const hasHistory = fullPrompt.includes('[Recent Context]'); console.log(`[jaw:pi] prompt len=${fullPrompt.length}, hasHistory=${hasHistory}, effort=${options.effort || 'none'}, sessionId=${options.sessionId || 'new'}`); - write('prompt', { message: fullPrompt }); + promptId = write('prompt', { message: fullPrompt }); return { child, done }; } diff --git a/src/agent/runtime/pi-turn.ts b/src/agent/runtime/pi-turn.ts new file mode 100644 index 000000000..ab8562060 --- /dev/null +++ b/src/agent/runtime/pi-turn.ts @@ -0,0 +1,135 @@ +import type { RuntimeTurnOutcome } from '../../shared/runtime-contract.js'; +import { appendBoundedFullText } from '../events/fulltext-bound.js'; + +type RecordValue = Record; +const record = (value: unknown): RecordValue => value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as RecordValue : {}; + +function assistant(value: unknown): Omit & { text: string | null } | null { + const message = record(value); + if (message['role'] !== 'assistant') return null; + let text: string | null = null, tool = false, oversized = false; + for (const part of Array.isArray(message['content']) ? message['content'] : []) { + const block = record(part); + if (block['type'] === 'toolCall') tool = true; + if (block['type'] !== 'text' || typeof block['text'] !== 'string') continue; + const bounded = appendBoundedFullText(text ?? '', block['text']); + text = bounded.text; oversized ||= bounded.truncated; + } + const reason = message['stopReason']; + const status = reason === 'aborted' ? 'stopped' + : !oversized && (reason === 'stop' || reason === 'toolUse') ? 'done' : 'error'; + return { text, status, finalText: status === 'done' && reason === 'stop' && !tool ? text : null }; +} + +/** agent_settled entered upstream RPC in 0.80.4; willRetry existed before it. */ +export function piSupportsSettled(version: string): boolean { + const match = /^(?:pi\s+)?v?(\d+)\.(\d+)\.(\d+)\s*$/.exec(version.trim()); + if (!match) return false; + const [, major, minor, patch] = match; + return Number(major) > 0 || Number(minor) > 80 || (Number(minor) === 80 && Number(patch) >= 4); +} + +/** Per admitted RPC prompt; no journal, session IDs, raw snapshots or callbacks retained. */ +export class PiTurnAccumulator { + private partial = ''; + private current = ''; + private runText = ''; + private completed = 0; + private candidate: Omit = { status: 'done', finalText: null }; + private ended = false; + + constructor(private readonly settledProtocol: boolean) {} + + private append(text: string): string { + const previous = this.partial.length; + this.partial = appendBoundedFullText(this.partial, text).text; + this.runText = appendBoundedFullText(this.runText, text).text; + this.current = appendBoundedFullText(this.current, text).text; + return text.slice(0, this.partial.length - previous); + } + + private reconcile(text: string | null): string { + // Snapshots echo prior deltas. A changed snapshot still owns finalText, + // but must not rewrite or duplicate already accepted salvage bytes. + if (text === null || !text.startsWith(this.current)) return ''; + return this.append(text.slice(this.current.length)); + } + + observe(value: unknown): { text: string; done: boolean } { + if (this.ended) return { text: '', done: false }; + const row = record(value); + let text = ''; + if (row['type'] === 'agent_start') { + this.current = ''; this.runText = ''; this.completed = 0; + this.candidate = { status: 'done', finalText: null }; + } else if (row['type'] === 'message_start' && record(row['message'])['role'] === 'assistant') { + this.current = ''; + this.candidate = { status: 'error', finalText: null }; + } else if (row['type'] === 'message_update') { + const event = record(row['assistantMessageEvent']); + const role = record(row['message'])['role']; + if ((role === undefined || role === 'assistant') && event['type'] === 'text_delta' && typeof event['delta'] === 'string') { + text = this.append(event['delta']); + } + } else if (row['type'] === 'message_end') { + const message = assistant(row['message']); + if (message) { + text = this.reconcile(message.text); + this.candidate = { status: message.status, finalText: message.finalText }; + this.completed++; + this.current = ''; + } + } else if (row['type'] === 'agent_end') { + text = this.terminal(row['messages']); + this.ended = !this.settledProtocol && row['willRetry'] !== true; + } else if (row['type'] === 'agent_settled') { + this.ended = true; + } + return { text, done: this.ended }; + } + + private terminal(value: unknown): string { + let index = 0, added = ''; + // Without message_end boundaries, the aggregate snapshot echoes the + // low-level run's stream. Consume that prefix by offset, not text identity. + let streamed = this.completed === 0 ? this.runText : ''; + this.candidate = { status: 'done', finalText: null }; + for (const entry of Array.isArray(value) ? value : []) { + const message = assistant(entry); + if (!message) continue; + this.candidate = { status: message.status, finalText: message.finalText }; + if (index++ < this.completed) continue; + if (this.completed === 0) { + const snapshot = message.text ?? ''; + const consumed = Math.min(streamed.length, snapshot.length); + this.current = streamed.slice(0, consumed); + streamed = streamed.slice(consumed); + } + added = appendBoundedFullText(added, this.reconcile(message.text)).text; + this.current = ''; + } + this.completed = index; + return added; + } + + snapshot(status?: RuntimeTurnOutcome['status']): RuntimeTurnOutcome { + return { status: status ?? this.candidate.status, + finalText: status ? null : this.candidate.finalText, partialText: this.partial }; + } +} + +/** Only this local carrier can supply failure outcome; arbitrary Error fields cannot. */ +export class PiRuntimeError extends Error { + readonly runtimeOutcome: RuntimeTurnOutcome; + constructor(cause: Error, outcome: RuntimeTurnOutcome) { + super(cause.message, { cause }); + this.name = 'PiRuntimeError'; + this.runtimeOutcome = { ...outcome }; + } +} + +export function piFailureOutcome(error: unknown): RuntimeTurnOutcome | undefined { + if (!(error instanceof PiRuntimeError) || !Object.hasOwn(error, 'runtimeOutcome')) return undefined; + return { ...error.runtimeOutcome }; +} diff --git a/structure/str_func.md b/structure/str_func.md index a13b225a4..6c8a3dfd5 100644 --- a/structure/str_func.md +++ b/structure/str_func.md @@ -112,7 +112,7 @@ cli-jaw/ │ │ ├── agy-bootstrap.ts ← AGY bootstrap/context preparation helpers (237L) │ │ ├── agy-capabilities.ts ← AGY `--help`/`--version` capability probe + cached optional flag support map + legacy emit-all fallback marker (124L) │ │ ├── agy-transcript-watcher.ts ← AGY transcript/log watcher and session-id extraction support (291L) -│ │ ├── pi-runtime.ts ← Pi profile 정규화 + isolated `PI_CODING_AGENT_DIR` models/settings 생성 + `pi --offline --list-models` discovery + `pi --mode rpc` JSONL parser/spawner (820L) ✨ +│ │ ├── pi-runtime.ts ← Pi profile 정규화 + isolated `PI_CODING_AGENT_DIR` models/settings 생성 + `pi --offline --list-models` discovery + `pi --mode rpc` JSONL parser/spawner (856L) ✨ │ │ ├── lifecycle-handler.ts ← child lifecycle + fallback/retry + queue resume orchestration + clearEmployeeSession on resume failure + stale resume fresh retry + kickGoalContinuation export + clearGoalTimers + goal continuation boundary row (1398L) │ │ ├── jwc-runtime.ts ← resident/in-process JWC runtime bridge and event handling (222L) │ │ ├── kiro-auth.ts ← Kiro CLI auth store reader (resolveKiroDataPath, readKiroAuthFromStore, resolveKiroProfileArn, regionFromProfileArn, listKiroConversationIdsForCwd, resolveKiroSessionIdAfterSpawn, extractKiroSessionIdFromV2Store) (253L) diff --git a/tests/unit/pi-finality-runtime.test.ts b/tests/unit/pi-finality-runtime.test.ts new file mode 100644 index 000000000..c003afcde --- /dev/null +++ b/tests/unit/pi-finality-runtime.test.ts @@ -0,0 +1,206 @@ +import '../setup/isolated-home.ts'; +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import { chmodSync, mkdtempSync, writeFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { spawnPiRpc, spawnPersistentPiRpc, DEFAULT_PI_PROFILE, DEFAULT_PI_SETTINGS } from '../../src/agent/pi-runtime.ts'; +import { PiTurnAccumulator, PiRuntimeError, piFailureOutcome, piSupportsSettled } from '../../src/agent/runtime/pi-turn.ts'; +import { FULLTEXT_MAX_CHARS } from '../../src/agent/events/fulltext-bound.ts'; + +const root = mkdtempSync(join(tmpdir(), 'pi-finality-')); +const binary = join(root, 'pi.mjs'); +const eventsFile = join(root, 'events.json'); +writeFileSync(binary, `#!/usr/bin/env node +import { readFileSync } from 'node:fs'; +import readline from 'node:readline'; +if (process.argv.includes('--version')) { console.log(process.env.PI_FINALITY_VERSION || '0.83.0'); process.exit(0); } +const send = row => console.log(JSON.stringify(row)); +for await (const line of readline.createInterface({input:process.stdin})) { + const request = JSON.parse(line); + if (request.type === 'get_state') send({type:'response',command:'get_state',id:request.id,success:true,data:{sessionId:'private-fixture-session'}}); + if (request.type === 'prompt') { + const rows = JSON.parse(readFileSync(process.env.PI_FINALITY_EVENTS,'utf8')); + for (const row of rows) send(row.id === '$prompt' ? {...row,id:request.id} : row); + } + if (request.type === 'test_settle') send({type:'agent_settled'}); + if (request.type === 'abort') { + send({type:'response',command:'abort',id:request.id,success:true}); + send({type:'agent_end',messages:[{role:'assistant',content:[],stopReason:'aborted'}],willRetry:false}); + send({type:'agent_settled'}); + } +} +`); +chmodSync(binary, 0o755); +const prior = {bin:process.env.PI_CODING_AGENT_BIN,events:process.env.PI_FINALITY_EVENTS,version:process.env.PI_FINALITY_VERSION}; +process.env.PI_CODING_AGENT_BIN = binary; +process.env.PI_FINALITY_EVENTS = eventsFile; +test.after(() => { + for (const [key, value] of Object.entries({PI_CODING_AGENT_BIN:prior.bin,PI_FINALITY_EVENTS:prior.events,PI_FINALITY_VERSION:prior.version})) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + rmSync(root, {recursive:true,force:true}); +}); +const assistant = (text: string | null, stopReason = 'stop', tool = false) => ({role:'assistant',stopReason, + content:[...(text === null ? [] : [{type:'text',text}]), ...(tool ? [{type:'toolCall',id:'private-tool',name:'bash',arguments:{command:'printf fixture'}}] : [])]}); +const delta = (text: string) => ({type:'message_update',assistantMessageEvent:{type:'text_delta',delta:text}}); +const end = (messages: unknown[]) => ({type:'agent_end',messages,willRetry:false}); +const settled = {type:'agent_settled'}; +function configure(rows: unknown[], version = '0.83.0') { + writeFileSync(eventsFile, JSON.stringify(rows)); process.env.PI_FINALITY_VERSION = version; +} +async function direct(rows: unknown[], rawThrows = false) { + configure(rows); + const accepted: string[] = []; + const run = spawnPiRpc(DEFAULT_PI_PROFILE, DEFAULT_PI_SETTINGS, {prompt:'fixture',model:'fixture',cwd:root,root, + onEvent:event => {if(event.kind==='text') accepted.push(event.text);}, + onRawRecord:rawThrows ? () => {throw new Error('journal unavailable');} : undefined}); + const closed = once(run.child,'close'); + const timeout = setTimeout(() => run.child.kill('SIGTERM'), 5000); + try { const result = await run.done; await closed; return {result,accepted:accepted.join('')}; } + finally { clearTimeout(timeout); run.child.kill(); } +} +const pre = assistant('Starting the read-only probe.', 'toolUse', true); +const final = assistant('PI_ACTIVITY_DONE'); +const probe = [ + {type:'agent_start'}, {type:'message_start',message:pre}, delta('Starting the read-only probe.'), + {type:'message_end',message:pre}, {type:'turn_end',message:pre,toolResults:[]}, + {type:'message_start',message:final},delta('PI_ACTIVITY_DONE'),{type:'message_end',message:final}, + {type:'turn_end',message:final,toolResults:[]},end([pre,final]),settled, +]; + +test('typed Pi finality excludes pre-tool commentary and preserves accepted text once', async () => { + const {result,accepted} = await direct(probe); + assert.deepEqual(Reflect.get(result,'runtimeOutcome'), {status:'done',finalText:'PI_ACTIVITY_DONE',partialText:'Starting the read-only probe.PI_ACTIVITY_DONE'}); + assert.equal(accepted,'Starting the read-only probe.PI_ACTIVITY_DONE'); +}); + +function accumulate(rows: unknown[], modern = true) { + const turn = new PiTurnAccumulator(modern); + const accepted: string[] = []; + for (const row of rows) accepted.push(turn.observe(row).text); + return {outcome:turn.snapshot(),accepted:accepted.join('')}; +} +for (const text of [null, '', ' \n\t ', 'FINAL']) { + test(`Pi final preserves ${JSON.stringify(text)} separately from pre-tool partial`, () => { + const result = accumulate([end([pre,assistant(text)]),settled]); + assert.deepEqual(result.outcome,{status:'done',finalText:text,partialText:pre.content[0]!.text + (text ?? '')}); + }); +} +for (const [reason,status] of [['error','error'],['aborted','stopped'],['length','error'],['pending','error'],['unknown','error'],['toolUse','done']]) { + test(`latest assistant ${reason} never promotes earlier completed answer`, () => { + const result = accumulate([end([assistant('old answer'),assistant('/goal done',reason)]),settled]); + assert.deepEqual(result.outcome,{status,finalText:null,partialText:'old answer/goal done'}); + }); +} +test('stop with toolCall and reasoning-only stop cannot manufacture a final', () => { + assert.equal(accumulate([end([assistant('commentary','stop',true)]),settled]).outcome.finalText,null); + const result = accumulate([end([{role:'assistant',stopReason:'stop',content:[{type:'thinking',thinking:'private reason'}]}]),settled]); + assert.deepEqual(result.outcome,{status:'done',finalText:null,partialText:''}); + assert.deepEqual(accumulate([end([{role:'toolResult',content:[{type:'text',text:'tool only'}]}]),settled]).outcome, + {status:'done',finalText:null,partialText:''}); +}); +test('distinct identical messages and mixed streamed/snapshot messages keep each occurrence once', () => { + const same = assistant('same'); + const result = accumulate([{type:'message_start',message:same},delta('sa'),{type:'message_end',message:same}, + {type:'turn_end',message:same},{type:'message_start',message:same},delta('s'),end([same,same]),settled]); + assert.equal(result.accepted,'samesame'); + assert.deepEqual(result.outcome,{status:'done',finalText:'same',partialText:'samesame'}); + assert.equal(accumulate([delta('same'),end([same,same]),end([same,same]),settled]).accepted,'samesame'); +}); +test('failure-only terminal retains earlier partial but invalidates prior successful candidate', () => { + const result = accumulate([{type:'message_start',message:final},delta('PI_ACTIVITY_DONE'), + {type:'message_end',message:final},end([assistant(null,'error')]),settled]); + assert.deepEqual(result.outcome,{status:'error',finalText:null,partialText:'PI_ACTIVITY_DONE'}); +}); +test('unknown control records and post-settlement events cannot rewrite a final', () => { + const result = accumulate([end([final]),settled,delta('late'),end([assistant('late')]), + {type:'mystery',text:'/goal done',messages:[assistant('wrong')]}]); + assert.deepEqual(result.outcome,{status:'done',finalText:'PI_ACTIVITY_DONE',partialText:'PI_ACTIVITY_DONE'}); +}); +test('partial cap survives oversized deltas and repeated terminal echoes', () => { + const text = 'x'.repeat(FULLTEXT_MAX_CHARS+1); + const result = accumulate([delta(text),end([assistant(text)]),end([assistant(text)]),settled]); + assert.equal(result.accepted.length,FULLTEXT_MAX_CHARS); + assert.equal(result.outcome.partialText.length,FULLTEXT_MAX_CHARS); + assert.equal(result.outcome.finalText,null,'oversized final is not silently truncated into success'); + assert.equal(result.outcome.status,'error'); +}); +test('modern retry/followup keeps overall salvage and takes the final low-level run', () => { + const turn = new PiTurnAccumulator(true); + turn.observe({type:'agent_start'}); turn.observe(delta('failed attempt')); + assert.equal(turn.observe({...end([assistant('failed attempt','error')]),willRetry:true}).done,false); + turn.observe({type:'agent_start'}); turn.observe(delta('intermediate')); + assert.equal(turn.observe(end([assistant('intermediate')])).done,false); + turn.observe({type:'agent_start'}); turn.observe(delta('final followup')); + turn.observe(end([assistant('final followup')])); + assert.equal(turn.observe(settled).done,true); + assert.deepEqual(turn.snapshot(),{status:'done',finalText:'final followup',partialText:'failed attemptintermediatefinal followup'}); +}); +test('settled capability uses the verified version boundary, not willRetry presence', () => { + for(const version of ['0.80.4','0.83.0','pi v0.81.0','1.0.0']) assert.equal(piSupportsSettled(version),true,version); + for(const version of ['0.75.4','0.80.3','0.80.4-beta','fake-pi 1.0.0','unknown']) assert.equal(piSupportsSettled(version),false,version); + const old = new PiTurnAccumulator(false); + assert.equal(old.observe({...end([assistant('retry','error')]),willRetry:true}).done,false); + old.observe({type:'agent_start'}); + assert.equal(old.observe(end([final])).done,true); +}); +test('failure carrier accepts only local owned snapshots', () => { + const original = new Error('original'); + const outcome = {status:'error' as const,finalText:null,partialText:'partial'}; + const error = new PiRuntimeError(original,outcome); outcome.partialText = 'mutated'; + assert.equal(error.cause,original); + assert.equal(piFailureOutcome(error)?.partialText,'partial'); + assert.equal(piFailureOutcome(Object.assign(new Error('foreign'),{runtimeOutcome:outcome})),undefined); + assert.equal(piFailureOutcome(Object.create(error)),undefined); +}); +test('raw observer failure leaves final and salvage intact', async context => { + context.mock.method(console,'warn',() => {}); + const {result,accepted} = await direct(probe,true); + assert.equal(result.runtimeOutcome?.finalText,'PI_ACTIVITY_DONE'); + assert.equal(result.runtimeOutcome?.partialText,accepted); +}); +test('modern persistent prompt remains active until explicit settled, then resets for reuse', async () => { + configure([end([pre,final])]); + const session = spawnPersistentPiRpc(DEFAULT_PI_PROFILE,DEFAULT_PI_SETTINGS,{model:'fixture',cwd:root,root}); + const closed = once(session.child,'close'); + let lowEnd!: () => void; + const ended = new Promise(resolve => {lowEnd=resolve;}); + let resolved = false; + const first = session.sendPrompt('first',{onRawRecord:row => {if(Reflect.get(row as object,'type')==='agent_end') lowEnd();}}); + void first.then(() => {resolved=true;}); + try { + await ended; await new Promise(resolve => setImmediate(resolve)); + assert.equal(resolved,false); + await assert.rejects(session.sendPrompt('overlap'),/already active/); + session.child.stdin!.write(JSON.stringify({type:'test_settle'})+'\n'); + assert.equal((await first).runtimeOutcome?.finalText,'PI_ACTIVITY_DONE'); + configure([end([assistant('second')]),settled]); + const second = await session.sendPrompt('second'); + assert.deepEqual(second.runtimeOutcome,{status:'done',finalText:'second',partialText:'second'}); + } finally { session.kill(); await closed; } +}); +test('pooled process termination rejects with bounded partial outcome', async () => { + configure([delta('interrupted partial')]); + const session = spawnPersistentPiRpc(DEFAULT_PI_PROFILE,DEFAULT_PI_SETTINGS,{model:'fixture',cwd:root,root}); + const closed = once(session.child,'close'); + const done = session.sendPrompt('hold',{onEvent:event => {if(event.kind==='text') session.kill();}}); + await assert.rejects(done,error => { + assert.deepEqual(piFailureOutcome(error),{status:'stopped',finalText:null,partialText:'interrupted partial'}); return true; + }); + await closed; +}); +test('correlated prompt rejection resolves direct error and rejects persistent with owned outcome', async () => { + const rows = [delta('accepted'),{type:'response',id:'$prompt',command:'prompt',success:false,error:'fixture rejection'}]; + assert.deepEqual((await direct(rows)).result.runtimeOutcome,{status:'error',finalText:null,partialText:'accepted'}); + configure(rows); + const session = spawnPersistentPiRpc(DEFAULT_PI_PROFILE,DEFAULT_PI_SETTINGS,{model:'fixture',cwd:root,root}); + const closed = once(session.child,'close'); + try { + await assert.rejects(session.sendPrompt('fixture'),error => { + assert.match(String(error),/fixture rejection/); + assert.deepEqual(piFailureOutcome(error),{status:'error',finalText:null,partialText:'accepted'}); return true; + }); + } finally {session.kill();await closed;} +}); From 9d14a20eb4113c55cad1023d1ac036d2e617e60c Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:16:20 +0900 Subject: [PATCH 03/12] fix(pi): hand off finality and salvage before lifecycle settlement --- src/agent/pi-runtime.ts | 12 ++- src/agent/runtime/pi-turn.ts | 6 +- src/agent/spawn.ts | 32 ++++-- structure/runtime-integration.md | 4 + structure/str_func.md | 5 +- tests/unit/pi-finality-runtime.test.ts | 14 ++- tests/unit/pi-finality-spawn.test.ts | 120 +++++++++++++++++++++ tests/unit/pi-spawn-runtime-events.test.ts | 30 +++++- 8 files changed, 201 insertions(+), 22 deletions(-) create mode 100644 tests/unit/pi-finality-spawn.test.ts diff --git a/src/agent/pi-runtime.ts b/src/agent/pi-runtime.ts index 322b55705..e39d8cece 100644 --- a/src/agent/pi-runtime.ts +++ b/src/agent/pi-runtime.ts @@ -283,20 +283,24 @@ export function resolvePiCommand(env: NodeJS.ProcessEnv = process.env): PiComman }; } -function resolvePiCommandIdentity(command: PiCommand, env: NodeJS.ProcessEnv = process.env): string { +function probePiCommandVersion(command: PiCommand, env: NodeJS.ProcessEnv) { const spec = launchSpec(command.command, [...command.baseArgs, '--version'], process.platform, env); - const result = spawnSync(spec.file, spec.args, { + return spawnSync(spec.file, spec.args, { encoding: 'utf8', env, timeout: 15_000, }); +} + +function resolvePiCommandIdentity(command: PiCommand, env: NodeJS.ProcessEnv = process.env): string { + const result = probePiCommandVersion(command, env); const version = `${result.stdout || ''}\n${result.stderr || ''}`.trim() || `exit:${String(result.status)}`; return JSON.stringify({ source: command.source, command: command.command, baseArgs: command.baseArgs, version }); } function usesPiSettled(command: PiCommand): boolean { - const identity = JSON.parse(resolvePiCommandIdentity(command)) as { version: string }; - return piSupportsSettled(identity.version); + const result = probePiCommandVersion(command, process.env); + return result.status === 0 && piSupportsSettled(result.stdout || ''); } function loadPiAbortEffective(profileId: string, command: PiCommand): boolean { diff --git a/src/agent/runtime/pi-turn.ts b/src/agent/runtime/pi-turn.ts index ab8562060..5ad903bd2 100644 --- a/src/agent/runtime/pi-turn.ts +++ b/src/agent/runtime/pi-turn.ts @@ -90,12 +90,16 @@ export class PiTurnAccumulator { } private terminal(value: unknown): string { + if (!Array.isArray(value)) { + this.candidate = { status: this.candidate.status === 'stopped' ? 'stopped' : 'error', finalText: null }; + return ''; + } let index = 0, added = ''; // Without message_end boundaries, the aggregate snapshot echoes the // low-level run's stream. Consume that prefix by offset, not text identity. let streamed = this.completed === 0 ? this.runText : ''; this.candidate = { status: 'done', finalText: null }; - for (const entry of Array.isArray(value) ? value : []) { + for (const entry of value) { const message = assistant(entry); if (!message) continue; this.candidate = { status: message.status, finalText: message.finalText }; diff --git a/src/agent/spawn.ts b/src/agent/spawn.ts index de7aaa255..719458fea 100644 --- a/src/agent/spawn.ts +++ b/src/agent/spawn.ts @@ -118,6 +118,8 @@ import { } from './kiro-runtime.js'; import { resolveCursorModelVariant } from './cursor-runtime.js'; import { normalizePiSettings, spawnPiRpc } from './pi-runtime.js'; +import { piFailureOutcome } from './runtime/pi-turn.js'; +import { handoffRuntimeOutcome } from './runtime/outcome.js'; import { getEmployeeMcpServers } from './mcp-passthrough.js'; // ─── State ─────────────────────────────────────────── @@ -2121,7 +2123,7 @@ export function spawnAgent(prompt: string, opts: SpawnOpts = {}): SpawnResult { } if (event.kind === 'session') ctx.sessionId = event.sessionId; }; - type PiTurnResult = { text: string; stderr: string; code: number; sessionId?: string | null }; + type PiTurnResult = { text: string; stderr: string; code: number; sessionId?: string | null; runtimeOutcome?: RuntimeTurnOutcome }; const runPiTurn = (child: ChildProcess, done: Promise, lease: PiLease | null): void => { let leaseCancel: Promise | null = null; const requestCancel = (): Promise => { @@ -2166,6 +2168,7 @@ export function spawnAgent(prompt: string, opts: SpawnOpts = {}): SpawnResult { else cleanupEmployeeTmpDir(spawnCwd, settings["workingDir"], agentLabel); }; done.then(async (result) => { + if (result.runtimeOutcome !== undefined) handoffRuntimeOutcome(ctx, result.runtimeOutcome); piWatchdog.stop(); await releaseLease(); flushPiThinking(); @@ -2199,18 +2202,23 @@ export function spawnAgent(prompt: string, opts: SpawnOpts = {}): SpawnResult { fallbackState: queueCtrl.fallbackStateForScope(scopeKey), fallbackMaxRetries: FALLBACK_MAX_RETRIES, processQueue, - }).finally(() => settleExit(scopeKey)); - }).catch(async (err: Error) => { + }); + }, async (err: Error) => { + const failedOutcome = piFailureOutcome(err); + if (failedOutcome !== undefined) handoffRuntimeOutcome(ctx, failedOutcome); + const killReason = consumeKillReason(child.pid); + const wasKilled = !!killReason; + const wasSteer = killReason === 'steer' || killReason === DUP_REGISTRATION_KILL_REASON; piWatchdog.stop(); await releaseLease().catch(() => {}); if (ctx.stderrBuf.length < 4000) ctx.stderrBuf += err.message; console.error('[jaw:pi] runtime failed:', err.message); - handleAgentExit({ + return handleAgentExit({ onRuntimeEnd: (end) => { activity.close(end); }, ctx, code: 1, cli, model: runtimeModel, effectiveProvider: profile.id, agentLabel, mainManaged, origin, resumeKey, prompt, opts, cfg, ownerGeneration, persistenceOwner, forceNew, empSid, - isResume: false, wasKilled: false, wasSteer: false, smokeResult: detectSmokeResponse('', [], 1, cli), + isResume: false, wasKilled, wasSteer, smokeResult: detectSmokeResponse('', [], 1, cli), effortDefault: 'medium', costLine: '', resolve: resolve!, activeProcesses, @@ -2224,11 +2232,15 @@ export function spawnAgent(prompt: string, opts: SpawnOpts = {}): SpawnResult { fallbackState: queueCtrl.fallbackStateForScope(scopeKey), fallbackMaxRetries: FALLBACK_MAX_RETRIES, processQueue, - }).catch((handleErr: Error) => { - activity.close({ kind: 'turn-end', status: 'error', finalText: null, error: 'Lifecycle failed' }); - console.error('[jaw:lifecycle] handleAgentExit failed (Pi):', handleErr.message); - }).finally(() => settleExit(scopeKey)); - }); + }); + }).catch((handleErr: Error) => { + // A lifecycle failure cannot re-enter delivery or become a new + // provider failure. Promise resolution is idempotent if delivered. + activity.close({ kind: 'turn-end', status: 'error', finalText: null, error: 'Lifecycle failed' }); + console.error('[jaw:lifecycle] handleAgentExit failed (Pi):', handleErr.message); + resolve!({ text: ctx.runtimeOutcome?.finalText ?? '', code: 1, + ...(ctx.runtimeOutcome === undefined ? {} : { runtimeOutcome: { ...ctx.runtimeOutcome } }) }); + }).finally(() => settleExit(scopeKey)); }; if (opts.agentId) { diff --git a/structure/runtime-integration.md b/structure/runtime-integration.md index a19d68320..3f18d1c81 100644 --- a/structure/runtime-integration.md +++ b/structure/runtime-integration.md @@ -8,6 +8,10 @@ tags: [cli-jaw, codex-app, pi, opencodex, runtime-pool] ## Shared event contract foundation +Pi final selection is owned by `src/agent/runtime/pi-turn.ts`, independently of Activity storage. The latest typed assistant message supplies a final only when its `stopReason` is `stop` and its content has no `toolCall`; text blocks preserve null, empty and whitespace distinctly. Earlier assistant commentary and accepted deltas stay in bounded `partialText`, while repeated message/turn/agent snapshots do not duplicate them. Error, abort, length and unknown stop reasons cannot promote an earlier answer. Pi invokes the existing explicit-outcome lifecycle handoff; adapters that omit an outcome retain legacy selection. + +Upstream Pi versions from 0.80.4 finish at `agent_settled`, after automatic retry, compaction and queued continuations. Older or unrecognized version strings retain the legacy `agent_end` boundary, except `willRetry:true`; this compatibility path cannot promise session-level settlement. `willRetry` itself predates `agent_settled` and is not a capability flag. Version probing is bounded and writes no shared capability/profile settings. Activity message phases may remain `unknown`; UI and journal replay never select the final. + `src/shared/runtime-contract.ts` defines native/print capabilities, distinct native-input/cancel-reprompt/queued/restart controls, and versioned presentation events. A jaw chat session and routing scope are separate from private provider session IDs. `RuntimeTurnOutcome` keeps authoritative `finalText` (null means absent; an empty string is intentional) separate from partial text. `src/agent/runtime/events.ts` records a validated, redacted body through the existing trace writer before publishing `agent_runtime` on the agent event topic. The trace writer owns sequence allocation; sequence gaps are valid. The tuple codec in `src/trace/runtime-body-codec.ts` preserves numeric usage without weakening raw-trace secret masking. Known structured fragments must be sanitized before clipping by their producer. Recording failure returns null, never a fabricated event or another inference. diff --git a/structure/str_func.md b/structure/str_func.md index 6c8a3dfd5..b123b4ec0 100644 --- a/structure/str_func.md +++ b/structure/str_func.md @@ -83,13 +83,14 @@ cli-jaw/ │ │ │ │ └── callbacks.ts ← bounded permission wait, cancellation latches and private replies (184L) │ │ │ ├── requests.ts ← ephemeral exact-bound decision registry and safe-view admission (158L) │ │ │ ├── pi-projection.ts ← Pi raw tool snapshots and accepted text/reasoning projection (97L) +│ │ │ ├── pi-turn.ts ← typed final selection, bounded partial and settlement compatibility (139L) │ │ │ ├── pi-raw-trace.ts ← bounded delta-only raw retention with explicit control summaries (81L) │ │ │ ├── projection.ts ← bounded redaction-before-clip snapshots and per-run failure latch (231L) │ │ │ ├── codex-projection.ts ← owned Codex notification mapping (98L) │ │ │ ├── outcome.ts ← non-journal native result handoff and stop precedence (31L) │ │ │ ├── session.ts ← native session/turn/control port (23L) │ │ │ └── events.ts ← validated trace-first semantic emitter (39L) -│ │ ├── spawn.ts ← CLI spawn + ACP/Codex App/Pi RPC/AGY/Kiro plain text/log session capture/claude-e helper 분기 + v2 SQLite session resume + 큐 + 메모리 flush + 429 retry timer + isAgentBusy/isSteerInProgress + buildHistoryBlock compact cutoff + working_dir scoping + enqueue→processQueue race fix + QueueItem persistent DB queue + makeCleanEnv PATH augment (3593L) +│ │ ├── spawn.ts ← CLI spawn + ACP/Codex App/Pi RPC/AGY/Kiro plain text/log session capture/claude-e helper 분기 + v2 SQLite session resume + 큐 + 메모리 flush + 429 retry timer + isAgentBusy/isSteerInProgress + buildHistoryBlock compact cutoff + working_dir scoping + enqueue→processQueue race fix + QueueItem persistent DB queue + makeCleanEnv PATH augment (3605L) │ │ ├── spawn/ ← spawn 서브모듈 (3 files) │ │ │ ├── queue.ts ← QueueItem persistent DB queue + processQueue race fix + enqueue/dequeue + drainRecoveredQueue (부팅 시 복구 큐 기동, server.ts가 transport 준비 후 호출) + `_fromQueue` 표식 (대기자 없는 턴을 채널이 답할 수 있게) (689L) │ │ │ ├── resume.ts ← session resume logic + stale resume detection (117L) @@ -112,7 +113,7 @@ cli-jaw/ │ │ ├── agy-bootstrap.ts ← AGY bootstrap/context preparation helpers (237L) │ │ ├── agy-capabilities.ts ← AGY `--help`/`--version` capability probe + cached optional flag support map + legacy emit-all fallback marker (124L) │ │ ├── agy-transcript-watcher.ts ← AGY transcript/log watcher and session-id extraction support (291L) -│ │ ├── pi-runtime.ts ← Pi profile 정규화 + isolated `PI_CODING_AGENT_DIR` models/settings 생성 + `pi --offline --list-models` discovery + `pi --mode rpc` JSONL parser/spawner (856L) ✨ +│ │ ├── pi-runtime.ts ← Pi profile 정규화 + isolated `PI_CODING_AGENT_DIR` models/settings 생성 + `pi --offline --list-models` discovery + `pi --mode rpc` JSONL parser/spawner (860L) ✨ │ │ ├── lifecycle-handler.ts ← child lifecycle + fallback/retry + queue resume orchestration + clearEmployeeSession on resume failure + stale resume fresh retry + kickGoalContinuation export + clearGoalTimers + goal continuation boundary row (1398L) │ │ ├── jwc-runtime.ts ← resident/in-process JWC runtime bridge and event handling (222L) │ │ ├── kiro-auth.ts ← Kiro CLI auth store reader (resolveKiroDataPath, readKiroAuthFromStore, resolveKiroProfileArn, regionFromProfileArn, listKiroConversationIdsForCwd, resolveKiroSessionIdAfterSpawn, extractKiroSessionIdFromV2Store) (253L) diff --git a/tests/unit/pi-finality-runtime.test.ts b/tests/unit/pi-finality-runtime.test.ts index c003afcde..f8bb3e080 100644 --- a/tests/unit/pi-finality-runtime.test.ts +++ b/tests/unit/pi-finality-runtime.test.ts @@ -15,7 +15,7 @@ const eventsFile = join(root, 'events.json'); writeFileSync(binary, `#!/usr/bin/env node import { readFileSync } from 'node:fs'; import readline from 'node:readline'; -if (process.argv.includes('--version')) { console.log(process.env.PI_FINALITY_VERSION || '0.83.0'); process.exit(0); } +if (process.argv.includes('--version')) { console.log(process.env.PI_FINALITY_VERSION || '0.83.0'); if(process.env.PI_FINALITY_WARNING) console.error('fixture warning'); process.exit(0); } const send = row => console.log(JSON.stringify(row)); for await (const line of readline.createInterface({input:process.stdin})) { const request = JSON.parse(line); @@ -119,6 +119,13 @@ test('unknown control records and post-settlement events cannot rewrite a final' {type:'mystery',text:'/goal done',messages:[assistant('wrong')]}]); assert.deepEqual(result.outcome,{status:'done',finalText:'PI_ACTIVITY_DONE',partialText:'PI_ACTIVITY_DONE'}); }); +test('malformed agent_end cannot mask an explicit error or stopped outcome', () => { + for (const [reason,status] of [['error','error'],['aborted','stopped']]) { + const result=accumulate([{type:'message_end',message:assistant('partial',reason)}, + {type:'agent_end',messages:'invalid'},settled]); + assert.deepEqual(result.outcome,{status,finalText:null,partialText:'partial'}); + } +}); test('partial cap survives oversized deltas and repeated terminal echoes', () => { const text = 'x'.repeat(FULLTEXT_MAX_CHARS+1); const result = accumulate([delta(text),end([assistant(text)]),end([assistant(text)]),settled]); @@ -161,7 +168,8 @@ test('raw observer failure leaves final and salvage intact', async context => { assert.equal(result.runtimeOutcome?.finalText,'PI_ACTIVITY_DONE'); assert.equal(result.runtimeOutcome?.partialText,accepted); }); -test('modern persistent prompt remains active until explicit settled, then resets for reuse', async () => { +for (const warning of [false,true]) test(`modern persistent prompt waits for settled despite version stderr warning=${warning}`, async () => { + if(warning) process.env.PI_FINALITY_WARNING='1'; configure([end([pre,final])]); const session = spawnPersistentPiRpc(DEFAULT_PI_PROFILE,DEFAULT_PI_SETTINGS,{model:'fixture',cwd:root,root}); const closed = once(session.child,'close'); @@ -179,7 +187,7 @@ test('modern persistent prompt remains active until explicit settled, then reset configure([end([assistant('second')]),settled]); const second = await session.sendPrompt('second'); assert.deepEqual(second.runtimeOutcome,{status:'done',finalText:'second',partialText:'second'}); - } finally { session.kill(); await closed; } + } finally { session.kill(); await closed; delete process.env.PI_FINALITY_WARNING; } }); test('pooled process termination rejects with bounded partial outcome', async () => { configure([delta('interrupted partial')]); diff --git a/tests/unit/pi-finality-spawn.test.ts b/tests/unit/pi-finality-spawn.test.ts new file mode 100644 index 000000000..6ebfb3524 --- /dev/null +++ b/tests/unit/pi-finality-spawn.test.ts @@ -0,0 +1,120 @@ +import '../setup/isolated-home.ts'; +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import { chmodSync, mkdirSync, mkdtempSync, writeFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { PiRpcSession } from '../../src/agent/pi-runtime.ts'; + +const root = mkdtempSync(join(tmpdir(),'pi-spawn-finality-')); +const binary = join(root,'pi.mjs'); +writeFileSync(binary, `#!/usr/bin/env node +import readline from 'node:readline'; +if(process.argv.includes('--version')) {console.log('0.83.0');process.exit(0);} +const send = row => console.log(JSON.stringify(row)); +for await(const line of readline.createInterface({input:process.stdin})) { + const r = JSON.parse(line); + if(r.type==='get_state') send({id:r.id,type:'response',command:r.type,success:true,data:{sessionId:'private-session'}}); + if(r.type==='prompt') { + send({type:'agent_start'}); + send({type:'message_update',assistantMessageEvent:{type:'text_delta',delta:'PROVISIONAL /goal done'}}); + if(process.env.PI_SPAWN_HOLD==='1') continue; + send({type:'agent_end',willRetry:false,messages:[{role:'assistant',stopReason:'toolUse',content:[{type:'text',text:'PROVISIONAL /goal done'}]}, + {role:'assistant',stopReason:'stop',content:[{type:'text',text:'FINAL ONLY'}]}]}); + send({type:'agent_settled'}); + } +} +`); +chmodSync(binary,0o755); +const previousBin = process.env.PI_CODING_AGENT_BIN; +process.env.PI_CODING_AGENT_BIN = binary; +const config = await import('../../src/core/config.ts'); +test.mock.module('../../src/core/config.js',{namedExports:{...config,detectCli:() => ({available:true,path:null})}}); +const pi = await import('../../src/agent/pi-runtime.ts'); +const sessions: PiRpcSession[] = []; +let onText: (() => void) | undefined; +test.mock.module('../../src/agent/pi-runtime.js',{namedExports:{...pi, + spawnPersistentPiRpc:(...args: Parameters) => { + const session = pi.spawnPersistentPiRpc(...args); + const send = session.sendPrompt.bind(session); + session.sendPrompt = (message,opts) => send(message,{...opts,onEvent:event => { + opts?.onEvent?.(event); if(event.kind==='text') onText?.(); + }}); + sessions.push(session); return session; + }, +}}); +const trace = await import('../../src/trace/store.ts'); +let failJournal = false; +test.mock.module('../../src/trace/store.js',{namedExports:{...trace, + appendTraceEvent:(...args: Parameters) => { + if(failJournal) throw new Error('fixture journal failed'); return trace.appendTraceEvent(...args); + }, +}}); +const {spawnAgent,killActiveAgent,waitForExitSettled,activeMainProcesses} = await import('../../src/agent/spawn.ts'); +const {db,getMaxMessageId,getSteerSalvageAfter} = await import('../../src/core/db.ts'); +const {subscribe} = await import('../../src/core/event-bus.ts'); +const {clearGoalTimers} = await import('../../src/agent/lifecycle-handler.ts'); +const {poolStats} = await import('../../src/agent/runtime-pool.ts'); +let serial = 0; +test.beforeEach(context => { + failJournal=false;onText=undefined;delete process.env.PI_SPAWN_HOLD; + config.settings.workingDir = root;mkdirSync(join(root,'prompts'),{recursive:true}); + mkdirSync(join(config.JAW_HOME,'prompts'),{recursive:true}); + config.settings.fallbackOrder=[];config.settings.activeOverrides={}; + config.settings.pi=pi.normalizePiSettings(pi.DEFAULT_PI_SETTINGS); + config.settings.perCli={...config.settings.perCli,pi:{model:'fixture',effort:'high',provider:'progrok'}}; + config.settings.memory={...config.settings.memory,enabled:false}; + config.settings.multiSession={enabled:true,maxConcurrent:4,midRunPolicy:'steer',channels:{telegram:true,discord:true,slack:true}}; + context.mock.method(globalThis,'fetch',async () => {throw new Error('unexpected network');}); + context.mock.method(console,'log',() => {});context.mock.method(console,'warn',() => {});context.mock.method(console,'error',() => {}); +}); +test.afterEach(async () => { + onText=undefined;clearGoalTimers(); + for(const session of sessions.splice(0)) { + if(session.child.exitCode!==null || session.child.signalCode!==null) continue; + const closed=once(session.child,'close');session.kill();await closed; + } + assert.equal(poolStats().busy,0); +}); +test.after(() => { + if(previousBin===undefined) delete process.env.PI_CODING_AGENT_BIN;else process.env.PI_CODING_AGENT_BIN=previousBin; + delete process.env.PI_SPAWN_HOLD;rmSync(root,{recursive:true,force:true}); +}); +function options() { + const id=++serial; + return {cli:'pi',model:'fixture',effort:'high',scopeKey:'pi-final-scope-'+id,chatSessionId:'pi-final-chat-'+id, + requestId:'pi-final-request-'+id,origin:'web',sysPrompt:'',_skipInsert:true,_skipHistory:true,_skipResume:true, + _skipSessionPersist:true,_isSmokeContinuation:true}; +} +test('actual pooled Pi-to-lifecycle final uses only typed final and canonical jaw identity',async () => { + const opts=options();const events: Record[]=[]; + const unsub=subscribe(event => {if(event.event==='agent_runtime') events.push(event.data as Record);}); + try { + const result=await spawnAgent('fixture',opts).promise; + assert.equal(result.text,'FINAL ONLY'); + assert.deepEqual(result.runtimeOutcome,{status:'done',finalText:'FINAL ONLY',partialText:'PROVISIONAL /goal doneFINAL ONLY'}); + const rows=db.prepare('SELECT content FROM messages WHERE session_id=? AND role=?').all(opts.chatSessionId,'assistant'); + assert.deepEqual(rows,[{content:'FINAL ONLY'}]); + const ends=events.filter(event => event.kind==='turn-end'); + assert.equal(ends.length,1);assert.equal(ends[0]?.finalText,'FINAL ONLY'); + assert.ok(events.every(event => event.sessionId===opts.chatSessionId && event.scope===opts.scopeKey)); + assert.doesNotMatch(JSON.stringify(events),/private-session/); + } finally {unsub();} +}); +test('kill-steered Pi rejection stores interrupted MESSAGE before the real exit barrier despite journal failure',async () => { + process.env.PI_SPAWN_HOLD='1';failJournal=true; + const opts=options();const watermark=getMaxMessageId(opts.chatSessionId); + let barrier:Promise|undefined;let observed:string|null|undefined; + onText=() => { + onText=undefined; + assert.equal(killActiveAgent(opts.scopeKey,'steer'),true); + barrier=waitForExitSettled(opts.scopeKey).then(() => {observed=getSteerSalvageAfter(opts.chatSessionId,watermark);}); + }; + const result=await spawnAgent('hold',opts).promise; + assert.ok(barrier,'real text callback armed kill-steer and exit barrier');await barrier; + assert.deepEqual(result.runtimeOutcome,{status:'stopped',finalText:null,partialText:'PROVISIONAL /goal done'}); + assert.equal(observed,'⏹️ [interrupted]\n\nPROVISIONAL /goal done'); + assert.equal(result.text,'');assert.notEqual(result.code,0); + assert.equal(activeMainProcesses.has(opts.scopeKey),false); +}); diff --git a/tests/unit/pi-spawn-runtime-events.test.ts b/tests/unit/pi-spawn-runtime-events.test.ts index c624f75cc..f226adec7 100644 --- a/tests/unit/pi-spawn-runtime-events.test.ts +++ b/tests/unit/pi-spawn-runtime-events.test.ts @@ -12,7 +12,7 @@ import type { RuntimeEventContext } from '../../src/agent/runtime/events.ts'; type Callbacks = { onEvent?: (event: PiRuntimeEvent) => void; onRawRecord?: (record: unknown) => void; cwd?: string }; const fixture = { - mode: 'ok' as 'ok' | 'acquire-failure' | 'direct-failure' | 'turn-failure' | 'raw-limit', + mode: 'ok' as 'ok' | 'acquire-failure' | 'direct-failure' | 'turn-failure' | 'raw-limit' | 'lifecycle-failure' | 'turn-lifecycle-failure', calls: [] as Callbacks[], acquisitions: [] as Array>, direct: 0, releases: 0, watchdogStops: 0, acquireGate: null as Promise | null, @@ -46,7 +46,7 @@ function child(): ChildProcess { } async function protocol(callbacks: Callbacks) { fixture.calls.push(callbacks); - if (fixture.mode === 'turn-failure') throw new Error('fixture Pi turn failed'); + if (fixture.mode === 'turn-failure' || fixture.mode === 'turn-lifecycle-failure') throw new Error('fixture Pi turn failed'); const raw = (record: unknown) => callbacks.onRawRecord?.(record); const semantic = (event: PiRuntimeEvent) => callbacks.onEvent?.(event); if (fixture.mode === 'raw-limit') raw({ type: 'fixture_oversize', payload: 'x'.repeat(70_000) }); @@ -109,7 +109,9 @@ test.mock.module('../../src/agent/lifecycle-handler.js', { namedExports: { params.releaseMainRun(params.scopeKey, params.childProcess, params.ownerGeneration); live.clearLiveRun(params.ctx.liveScope || 'default'); traces.finalizeTraceRun(params.ctx.traceRunId, params.code === 0 ? 'done' : 'error'); + if (fixture.mode === 'turn-lifecycle-failure') throw new Error('fixture failed lifecycle before caller resolution'); params.resolve({ text: finalText ?? '', code: params.code ?? 0, tools: params.ctx.toolLog }); + if (fixture.mode === 'lifecycle-failure') throw new Error('fixture failure after finalization'); }, } }); const { spawnAgent, activeProcesses, activeMainProcesses, armExitSettle, waitForExitSettled, settleExit } = await import('../../src/agent/spawn.ts'); @@ -212,6 +214,30 @@ test('rejected Pi turn uses error lifecycle observer once and releases pooled le assert.equal(activeMainProcesses.has('pi-test-scope'), false); }); +test('Pi lifecycle rejection after finalization cannot execute lifecycle or release twice', async () => { + fixture.mode = 'lifecycle-failure'; + const result = await spawnAgent('lifecycle fixture', opts()).promise; + await new Promise(resolve => setImmediate(resolve)); + assert.equal(result.text, 'lifecycle-selected final'); + assert.equal(fixture.lifecycle.length, 1); + assert.equal(fixture.releases, 1); + assert.equal(fixture.events.filter(event => event.kind === 'turn-end').length, 1); +}); +test('Pi execution failure followed by lifecycle rejection resolves caller once and settles barrier', async () => { + fixture.mode = 'turn-lifecycle-failure'; + const scope = 'pi-test-scope'; + armExitSettle(scope); + let barrierDone = false; + const barrier = waitForExitSettled(scope).then(() => { barrierDone = true; }); + const result = await spawnAgent('double failure fixture', opts()).promise; + await new Promise(resolve => setImmediate(resolve)); + assert.equal(result.code, 1); assert.equal(result.text, ''); + assert.equal(fixture.lifecycle.length, 1); assert.equal(fixture.releases, 1); + assert.equal(barrierDone, true); + assert.equal(fixture.events.filter(event => event.kind === 'turn-end').length, 1); + await barrier; +}); + test('Pi acquire failure closes trace/live state and settles the armed exit barrier without timeout', async context => { fixture.mode = 'acquire-failure'; context.mock.timers.enable({ apis: ['setTimeout'] }); From 49976855661ac04faf418ffa9e0e92adeca7fc52 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:17:29 +0900 Subject: [PATCH 04/12] docs: record Pi finality verification evidence --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 80b8290c9..dd780b02e 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 80b8290c960eb1f801f231a32c84f658041b15ae +Subproject commit dd780b02eaf610ba88f78e28372216959cab436c From c87e5be9b5029578b000b9ac454540428d78569c Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:18:47 +0900 Subject: [PATCH 05/12] test(pi): cover user stop and settled continuation boundaries --- tests/unit/pi-finality-runtime.test.ts | 18 ++++++++++++++++-- tests/unit/pi-finality-spawn.test.ts | 11 +++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/tests/unit/pi-finality-runtime.test.ts b/tests/unit/pi-finality-runtime.test.ts index f8bb3e080..ee0381bad 100644 --- a/tests/unit/pi-finality-runtime.test.ts +++ b/tests/unit/pi-finality-runtime.test.ts @@ -25,6 +25,7 @@ for await (const line of readline.createInterface({input:process.stdin})) { for (const row of rows) send(row.id === '$prompt' ? {...row,id:request.id} : row); } if (request.type === 'test_settle') send({type:'agent_settled'}); + if (request.type === 'test_events') for(const row of request.events) send(row); if (request.type === 'abort') { send({type:'response',command:'abort',id:request.id,success:true}); send({type:'agent_end',messages:[{role:'assistant',content:[],stopReason:'aborted'}],willRetry:false}); @@ -182,8 +183,10 @@ for (const warning of [false,true]) test(`modern persistent prompt waits for set await ended; await new Promise(resolve => setImmediate(resolve)); assert.equal(resolved,false); await assert.rejects(session.sendPrompt('overlap'),/already active/); - session.child.stdin!.write(JSON.stringify({type:'test_settle'})+'\n'); - assert.equal((await first).runtimeOutcome?.finalText,'PI_ACTIVITY_DONE'); + session.child.stdin!.write(JSON.stringify({type:'test_events',events:[{type:'agent_start'},end([assistant('queued continuation')]),settled]})+'\n'); + const completed=await first; + assert.equal(completed.runtimeOutcome?.finalText,'queued continuation'); + assert.equal(completed.runtimeOutcome?.partialText,'Starting the read-only probe.PI_ACTIVITY_DONEqueued continuation'); configure([end([assistant('second')]),settled]); const second = await session.sendPrompt('second'); assert.deepEqual(second.runtimeOutcome,{status:'done',finalText:'second',partialText:'second'}); @@ -199,6 +202,17 @@ test('pooled process termination rejects with bounded partial outcome', async () }); await closed; }); +test('modern correlated abort waits for terminal and preserves stopped partial', async () => { + configure([delta('before abort')]); + const session=spawnPersistentPiRpc(DEFAULT_PI_PROFILE,DEFAULT_PI_SETTINGS,{model:'fixture',cwd:root,root}); + const closed=once(session.child,'close'); + let aborting:Promise|undefined; + try { + const result=await session.sendPrompt('hold',{onEvent:event => {if(event.kind==='text') aborting=session.abort();}}); + assert.ok(aborting);await aborting; + assert.deepEqual(result.runtimeOutcome,{status:'stopped',finalText:null,partialText:'before abort'}); + } finally {session.kill();await closed;} +}); test('correlated prompt rejection resolves direct error and rejects persistent with owned outcome', async () => { const rows = [delta('accepted'),{type:'response',id:'$prompt',command:'prompt',success:false,error:'fixture rejection'}]; assert.deepEqual((await direct(rows)).result.runtimeOutcome,{status:'error',finalText:null,partialText:'accepted'}); diff --git a/tests/unit/pi-finality-spawn.test.ts b/tests/unit/pi-finality-spawn.test.ts index 6ebfb3524..aeebccbd4 100644 --- a/tests/unit/pi-finality-spawn.test.ts +++ b/tests/unit/pi-finality-spawn.test.ts @@ -118,3 +118,14 @@ test('kill-steered Pi rejection stores interrupted MESSAGE before the real exit assert.equal(result.text,'');assert.notEqual(result.code,0); assert.equal(activeMainProcesses.has(opts.scopeKey),false); }); + +test('user stop preserves partial outcome without inventing a final response',async () => { + process.env.PI_SPAWN_HOLD='1'; + const opts=options(); + onText=() => {onText=undefined;assert.equal(killActiveAgent(opts.scopeKey,'user'),true);}; + const result=await spawnAgent('hold',opts).promise; + assert.deepEqual(result.runtimeOutcome,{status:'stopped',finalText:null,partialText:'PROVISIONAL /goal done'}); + assert.equal(result.text,'');assert.notEqual(result.code,0); + assert.equal(activeMainProcesses.has(opts.scopeKey),false); + assert.deepEqual(db.prepare('SELECT content FROM messages WHERE session_id=? AND role=?').all(opts.chatSessionId,'assistant'),[]); +}); From d9f464e8415b06cfe3fc7463d917b975c683f179 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:25:16 +0900 Subject: [PATCH 06/12] fix(pi): preserve failure status and isolate exit observers --- src/agent/runtime/pi-turn.ts | 17 +++++++++++------ src/agent/spawn.ts | 3 ++- structure/str_func.md | 4 ++-- tests/unit/pi-finality-runtime.test.ts | 14 ++++++++++++++ tests/unit/pi-finality-spawn.test.ts | 8 ++++++++ 5 files changed, 37 insertions(+), 9 deletions(-) diff --git a/src/agent/runtime/pi-turn.ts b/src/agent/runtime/pi-turn.ts index 5ad903bd2..e8988b1fb 100644 --- a/src/agent/runtime/pi-turn.ts +++ b/src/agent/runtime/pi-turn.ts @@ -8,17 +8,20 @@ const record = (value: unknown): RecordValue => value !== null && typeof value = function assistant(value: unknown): Omit & { text: string | null } | null { const message = record(value); if (message['role'] !== 'assistant') return null; - let text: string | null = null, tool = false, oversized = false; - for (const part of Array.isArray(message['content']) ? message['content'] : []) { + const content = message['content']; + let text: string | null = null, tool = false, oversized = false, invalid = !Array.isArray(content); + for (const part of Array.isArray(content) ? content : []) { const block = record(part); - if (block['type'] === 'toolCall') tool = true; - if (block['type'] !== 'text' || typeof block['text'] !== 'string') continue; + if (block['type'] === 'toolCall') { tool = true; continue; } + if (block['type'] === 'thinking') continue; + if (block['type'] !== 'text' || typeof block['text'] !== 'string') { invalid = true; continue; } + if (block['text'] === '') { text ??= ''; continue; } const bounded = appendBoundedFullText(text ?? '', block['text']); text = bounded.text; oversized ||= bounded.truncated; } const reason = message['stopReason']; const status = reason === 'aborted' ? 'stopped' - : !oversized && (reason === 'stop' || reason === 'toolUse') ? 'done' : 'error'; + : !invalid && !oversized && (reason === 'stop' || reason === 'toolUse') ? 'done' : 'error'; return { text, status, finalText: status === 'done' && reason === 'stop' && !tool ? text : null }; } @@ -98,7 +101,9 @@ export class PiTurnAccumulator { // Without message_end boundaries, the aggregate snapshot echoes the // low-level run's stream. Consume that prefix by offset, not text identity. let streamed = this.completed === 0 ? this.runText : ''; - this.candidate = { status: 'done', finalText: null }; + // An empty/tool-only terminal cannot erase an observed failure. Only + // an actual assistant snapshot can supply a newer completion status. + this.candidate = { status: this.candidate.status, finalText: null }; for (const entry of value) { const message = assistant(entry); if (!message) continue; diff --git a/src/agent/spawn.ts b/src/agent/spawn.ts index 719458fea..1b8e3894c 100644 --- a/src/agent/spawn.ts +++ b/src/agent/spawn.ts @@ -2175,7 +2175,8 @@ export function spawnAgent(prompt: string, opts: SpawnOpts = {}): SpawnResult { if (ctx.stderrBuf.length < 4000) ctx.stderrBuf += result.stderr || ''; if (result.sessionId) ctx.sessionId = result.sessionId; if (!ctx.fullText && result.text) ctx.fullText = result.text; - opts.lifecycle?.onExit?.(result.code); + try { opts.lifecycle?.onExit?.(result.code); } + catch { console.warn('[jaw:pi] exit observer failed'); } const killReason = consumeKillReason(child.pid); const wasKilled = !!killReason; // 'dup-registration' behaves like a steer for cleanup purposes: a diff --git a/structure/str_func.md b/structure/str_func.md index b123b4ec0..4b61bdb50 100644 --- a/structure/str_func.md +++ b/structure/str_func.md @@ -83,14 +83,14 @@ cli-jaw/ │ │ │ │ └── callbacks.ts ← bounded permission wait, cancellation latches and private replies (184L) │ │ │ ├── requests.ts ← ephemeral exact-bound decision registry and safe-view admission (158L) │ │ │ ├── pi-projection.ts ← Pi raw tool snapshots and accepted text/reasoning projection (97L) -│ │ │ ├── pi-turn.ts ← typed final selection, bounded partial and settlement compatibility (139L) +│ │ │ ├── pi-turn.ts ← typed final selection, bounded partial and settlement compatibility (144L) │ │ │ ├── pi-raw-trace.ts ← bounded delta-only raw retention with explicit control summaries (81L) │ │ │ ├── projection.ts ← bounded redaction-before-clip snapshots and per-run failure latch (231L) │ │ │ ├── codex-projection.ts ← owned Codex notification mapping (98L) │ │ │ ├── outcome.ts ← non-journal native result handoff and stop precedence (31L) │ │ │ ├── session.ts ← native session/turn/control port (23L) │ │ │ └── events.ts ← validated trace-first semantic emitter (39L) -│ │ ├── spawn.ts ← CLI spawn + ACP/Codex App/Pi RPC/AGY/Kiro plain text/log session capture/claude-e helper 분기 + v2 SQLite session resume + 큐 + 메모리 flush + 429 retry timer + isAgentBusy/isSteerInProgress + buildHistoryBlock compact cutoff + working_dir scoping + enqueue→processQueue race fix + QueueItem persistent DB queue + makeCleanEnv PATH augment (3605L) +│ │ ├── spawn.ts ← CLI spawn + ACP/Codex App/Pi RPC/AGY/Kiro plain text/log session capture/claude-e helper 분기 + v2 SQLite session resume + 큐 + 메모리 flush + 429 retry timer + isAgentBusy/isSteerInProgress + buildHistoryBlock compact cutoff + working_dir scoping + enqueue→processQueue race fix + QueueItem persistent DB queue + makeCleanEnv PATH augment (3606L) │ │ ├── spawn/ ← spawn 서브모듈 (3 files) │ │ │ ├── queue.ts ← QueueItem persistent DB queue + processQueue race fix + enqueue/dequeue + drainRecoveredQueue (부팅 시 복구 큐 기동, server.ts가 transport 준비 후 호출) + `_fromQueue` 표식 (대기자 없는 턴을 채널이 답할 수 있게) (689L) │ │ │ ├── resume.ts ← session resume logic + stale resume detection (117L) diff --git a/tests/unit/pi-finality-runtime.test.ts b/tests/unit/pi-finality-runtime.test.ts index ee0381bad..d914f8eab 100644 --- a/tests/unit/pi-finality-runtime.test.ts +++ b/tests/unit/pi-finality-runtime.test.ts @@ -127,6 +127,20 @@ test('malformed agent_end cannot mask an explicit error or stopped outcome', () assert.deepEqual(result.outcome,{status,finalText:null,partialText:'partial'}); } }); +test('terminal arrays without an assistant cannot erase prior error or abort', () => { + for (const [reason,status] of [['error','error'],['aborted','stopped']]) { + for (const messages of [[],[null],[{role:'toolResult',content:[]}]]) { + const result=accumulate([{type:'message_end',message:assistant('partial',reason)},end(messages),settled]); + assert.deepEqual(result.outcome,{status,finalText:null,partialText:'partial'}); + } + } +}); +test('malformed assistant content cannot claim a successful typed final', () => { + for (const content of [null,'raw text',[{type:'text',text:1}],[{type:'future-control',text:'FINAL'}]]) { + const result=accumulate([end([{role:'assistant',stopReason:'stop',content}]),settled]); + assert.equal(result.outcome.status,'error');assert.equal(result.outcome.finalText,null); + } +}); test('partial cap survives oversized deltas and repeated terminal echoes', () => { const text = 'x'.repeat(FULLTEXT_MAX_CHARS+1); const result = accumulate([delta(text),end([assistant(text)]),end([assistant(text)]),settled]); diff --git a/tests/unit/pi-finality-spawn.test.ts b/tests/unit/pi-finality-spawn.test.ts index aeebccbd4..0eecef79c 100644 --- a/tests/unit/pi-finality-spawn.test.ts +++ b/tests/unit/pi-finality-spawn.test.ts @@ -102,6 +102,14 @@ test('actual pooled Pi-to-lifecycle final uses only typed final and canonical ja assert.doesNotMatch(JSON.stringify(events),/private-session/); } finally {unsub();} }); +test('throwing exit observer cannot bypass lifecycle cleanup or final MESSAGE',async () => { + const opts=options(); + const result=await spawnAgent('fixture',{...opts,lifecycle:{onExit:() => {throw new Error('fixture observer');}}}).promise; + assert.equal(result.text,'FINAL ONLY');assert.equal(result.code,0); + assert.equal(result.runtimeOutcome?.status,'done'); + assert.equal(activeMainProcesses.has(opts.scopeKey),false); + assert.deepEqual(db.prepare('SELECT content FROM messages WHERE session_id=? AND role=?').all(opts.chatSessionId,'assistant'),[{content:'FINAL ONLY'}]); +}); test('kill-steered Pi rejection stores interrupted MESSAGE before the real exit barrier despite journal failure',async () => { process.env.PI_SPAWN_HOLD='1';failJournal=true; const opts=options();const watermark=getMaxMessageId(opts.chatSessionId); From 897e3d35b25a238900b88fc091678fec15695e5b Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:27:48 +0900 Subject: [PATCH 07/12] docs: attach final Pi review evidence --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index dd780b02e..9dec23901 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit dd780b02eaf610ba88f78e28372216959cab436c +Subproject commit 9dec2390140f24467654947c4cd9c49569b831c2 From f3d1dc5dd2b1c638554a4e8e5b22075996bcb908 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:55:45 +0900 Subject: [PATCH 08/12] docs: align API inventory counts with inherited routes --- structure/server_api.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/structure/server_api.md b/structure/server_api.md index 1092f6d6f..db99b8a01 100644 --- a/structure/server_api.md +++ b/structure/server_api.md @@ -165,7 +165,7 @@ Cursor/Grok activation and Activity controls are separate from this API foundati | Dashboard Schedule | `GET /api/dashboard/schedule/work` `POST /api/dashboard/schedule/work` `PATCH /api/dashboard/schedule/work/:id` `DELETE /api/dashboard/schedule/work/:id` `POST /api/dashboard/schedule/work/:id/dispatch` | | i18n | `GET /api/i18n/languages` `GET /api/i18n/:lang` | -> 실제 코드(`server.ts` + `src/routes/*.ts` + mounted runtime/security/Jaw CEO/dashboard sub-router)에서 추출한 총 256개 route handler 기준이다. 이 중 API 엔드포인트는 255개이고, 나머지 1개는 `/` 엔트리이다. Browser API 43개는 `src/routes/browser.ts`에서 등록된다. Jaw CEO 20개는 `src/routes/jaw-ceo.ts`에서 sub-router로 등록된다. +> 실제 코드(`server.ts` + `src/routes/*.ts` + mounted runtime/security/Jaw CEO/dashboard sub-router)에서 추출한 총 258개 route handler 기준이다. 이 중 API 엔드포인트는 257개이고, 나머지 1개는 `/` 엔트리이다. Browser API 43개는 `src/routes/browser.ts`에서 등록된다. Jaw CEO 20개는 `src/routes/jaw-ceo.ts`에서 sub-router로 등록된다. `PUT /api/heartbeat`의 job은 `mentionWatch: { channel: "slack", userId: "U...", channelIds: ["C..."], maxHits?, since? }`를 선택적으로 받는다. `channelIds`는 비어 있지 않아야 하고 저장 시 `slack.channelIds` allowlist의 부분집합이어야 하며, 실행 tick 직전 현재 allowlist와 다시 교집합한다. job id가 같은 기존 값에 대해 필드가 없으면 상속하고, `null`이면 삭제하며, 잘못된 값은 `400 invalid heartbeat mention watch`다. 파일 로드 정규화에서 잘못된 `mentionWatch`는 해당 job을 `enabled: false`로 내린다. 기본 운영값은 비활성이고, 설정된 watch는 별도 daemon이 아니라 기존 `runHeartbeatJob`에서 실행된다. From 67100bd547df6c92358795c3cce07c8e010ba839 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 03:55:46 +0900 Subject: [PATCH 09/12] docs: attach CI baseline repair evidence --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 9dec23901..25fac368e 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 9dec2390140f24467654947c4cd9c49569b831c2 +Subproject commit 25fac368e3242caf3a815f23e22b0c981db3b194 From 0d74bbd27f5c8884ad4c3ef4df7cf906b50dd988 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 04:08:27 +0900 Subject: [PATCH 10/12] docs: close Pi finality implementation record --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 25fac368e..599c5a12c 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 25fac368e3242caf3a815f23e22b0c981db3b194 +Subproject commit 599c5a12c21de537456b8f53fa260d3c0157a199 From 5279a13ddf2bfec15f90fef0843023c1cf0e0ea2 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 07:49:35 +0900 Subject: [PATCH 11/12] docs: align private cascade ancestry --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 599c5a12c..9f91ec4ad 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 599c5a12c21de537456b8f53fa260d3c0157a199 +Subproject commit 9f91ec4adedcf4e0b6865a2371ec3b11d4f0ccce From fe7d6837983ed94ee8e966e4ea024528ea1b2bf6 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 6 Sep 2026 07:54:35 +0900 Subject: [PATCH 12/12] docs: record Pi cascade verification and parent ancestry --- devlog | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devlog b/devlog index 9f91ec4ad..cd6d62720 160000 --- a/devlog +++ b/devlog @@ -1 +1 @@ -Subproject commit 9f91ec4adedcf4e0b6865a2371ec3b11d4f0ccce +Subproject commit cd6d6272094a4ecb85faadc721a987fbb612e01c