test(coding-agent): implemented guest reconciliation and expanded testing for collaboration

- Introduced guest snapshot reconciliation to maintain host state consistency during session switching.
- Improved yield tool reliability by implementing incremental schema validation and strict parameter enforcement.
- Fixed a calculation edge case in the status line to prevent negative time values during activity tracking.
- Expanded the test suite with new validation for session interruption, collab state synchronization, and process error handling.
This commit is contained in:
can1357
2026-06-28 09:52:44 +02:00
parent 1190aade82
commit 51a2a0342f
10 changed files with 335 additions and 7 deletions
+14
View File
@@ -105,6 +105,19 @@ export function reconcileGuestIdleHostState(ctx: GuestIdleReconcilerCtx, isStrea
}
}
/** Reconcile a welcome/resync snapshot's host activity state into the guest meter. */
export interface GuestSnapshotActivityReconcilerCtx extends GuestIdleReconcilerCtx {
statusLine: GuestIdleReconcilerCtx["statusLine"] & { markActivityStart: () => void };
}
export function reconcileGuestSnapshotHostState(ctx: GuestSnapshotActivityReconcilerCtx, isStreaming: boolean): void {
if (isStreaming) {
ctx.statusLine.markActivityStart();
return;
}
reconcileGuestIdleHostState(ctx, false);
}
export class CollabGuestLink {
#ctx: InteractiveModeContext;
#socket: CollabSocket | null = null;
@@ -382,6 +395,7 @@ export class CollabGuestLink {
this.#clearAgentMirror();
await this.#ctx.session.switchSession(replicaPath);
this.state = pending.state;
reconcileGuestSnapshotHostState(this.#ctx, pending.state.isStreaming);
this.#applyHostState(pending.state);
this.#ctx.resetObserverRegistry();
this.#applyAgentSnapshots(pending.agents);
@@ -408,7 +408,7 @@ export class StatusLineComponent implements Component {
markActivityEnd(): void {
const meter = this.#meter();
if (meter.activeStartedAt === null) return;
meter.activeMs += Date.now() - meter.activeStartedAt;
meter.activeMs += Math.max(0, Date.now() - meter.activeStartedAt);
meter.activeStartedAt = null;
}
@@ -420,7 +420,7 @@ export class StatusLineComponent implements Component {
getActiveMs(): number {
const meter = this.#meter();
if (meter.activeStartedAt === null) return meter.activeMs;
return meter.activeMs + Date.now() - meter.activeStartedAt;
return meter.activeMs + Math.max(0, Date.now() - meter.activeStartedAt);
}
/**
@@ -1112,6 +1112,7 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor {
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
reasoningTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
};
let hasUsage = false;
@@ -1502,6 +1503,8 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor {
accumulatedUsage.cacheRead += getNumberField(usageRecord, "cacheRead") ?? 0;
accumulatedUsage.cacheWrite += getNumberField(usageRecord, "cacheWrite") ?? 0;
accumulatedUsage.totalTokens += getNumberField(usageRecord, "totalTokens") ?? 0;
accumulatedUsage.reasoningTokens =
(accumulatedUsage.reasoningTokens ?? 0) + (getNumberField(usageRecord, "reasoningTokens") ?? 0);
if (costRecord) {
accumulatedUsage.cost.input += getNumberField(costRecord, "input") ?? 0;
accumulatedUsage.cost.output += getNumberField(costRecord, "output") ?? 0;
+41 -1
View File
@@ -98,6 +98,46 @@ function parseYieldType(value: unknown): string | string[] | undefined {
throw new Error("type must be a string or non-empty array of strings");
}
/**
* Expand a plain-object `data` schema into a strict union that ALSO accepts each
* top-level section value (and array element) on its own. Agents that yield
* incrementally (`type: ["findings"]`, `type: ["confidence"]`, …) submit one
* section per call, so `data` is a single finding object or a lone verdict value
* — never the full output object. Without this, strict-mode providers constrain
* `data` to the whole schema and reject/—under constrained decoding—forbid the
* partial. Every branch is a typed sub-schema, so strict representability holds;
* the full-output object stays the first (terminal) branch. The assembled whole
* is still validated against the full schema at finalization. Non-object / loose
* schemas are returned unchanged.
*/
function withSectionVariants(dataSchema: Record<string, unknown>): Record<string, unknown> {
if (dataSchema.type !== "object") return dataSchema;
const props = dataSchema.properties;
if (props === null || typeof props !== "object") return dataSchema;
const propRecord = props as Record<string, unknown>;
const { description, ...fullWithoutDescription } = dataSchema;
const branches: unknown[] = [];
const seen = new Set<string>();
const add = (schema: unknown): void => {
if (schema === null || typeof schema !== "object") return;
const key = JSON.stringify(schema);
if (seen.has(key)) return;
seen.add(key);
branches.push(schema);
};
add(fullWithoutDescription);
for (const name in propRecord) {
const prop = propRecord[name];
add(prop);
if (prop !== null && typeof prop === "object") {
const propObj = prop as Record<string, unknown>;
if (propObj.type === "array") add(propObj.items);
}
}
if (branches.length <= 1) return dataSchema;
return description !== undefined ? { description, anyOf: branches } : { anyOf: branches };
}
function wrapYieldParameters(dataSchema: Record<string, unknown>): Record<string, unknown> {
const successResultSchema = {
type: "object",
@@ -206,7 +246,7 @@ export class YieldTool implements AgentTool<TSchema, YieldDetails> {
if (hasUnresolvedRefs(resolved)) {
throw new Error("schema contains unresolved $ref after dereferencing");
}
dataSchema = resolved;
dataSchema = withSectionVariants(resolved);
} else {
this.strict = false;
dataSchema = looseRecordSchema(
@@ -5,6 +5,8 @@ import type { Api, AssistantMessage, Model, ThinkingContent } from "@oh-my-pi/pi
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry";
import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings";
import { ExtensionRuntime, loadExtensionFromFactory } from "@oh-my-pi/pi-coding-agent/extensibility/extensions/loader";
import { ExtensionRunner } from "@oh-my-pi/pi-coding-agent/extensibility/extensions/runner";
import { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session";
import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage";
import {
@@ -15,6 +17,7 @@ import {
} from "@oh-my-pi/pi-coding-agent/session/messages";
import type { SessionEntry } from "@oh-my-pi/pi-coding-agent/session/session-entries";
import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
import { EventBus } from "@oh-my-pi/pi-coding-agent/utils/event-bus";
import { TempDir } from "@oh-my-pi/pi-utils";
const REASONING_TEXT = "I have partly reasoned through the implementation and should preserve this.";
@@ -111,7 +114,11 @@ describe("AgentSession interrupted thinking persistence", () => {
}
});
function createSession(): { model: Model<Api>; sessionManager: SessionManager; session: AgentSession } {
function createSession(extensionRunner?: ExtensionRunner): {
model: Model<Api>;
sessionManager: SessionManager;
session: AgentSession;
} {
const model = getBundledModel("anthropic", "claude-sonnet-4-5");
const agent = new Agent({
getApiKey: () => "anthropic-test-key",
@@ -131,6 +138,7 @@ describe("AgentSession interrupted thinking persistence", () => {
sessionManager,
settings,
modelRegistry: new ModelRegistry(authStorage, path.join(tempDir.path(), "models.yml")),
extensionRunner,
});
return { model, sessionManager, session };
}
@@ -197,6 +205,50 @@ describe("AgentSession interrupted thinking persistence", () => {
expect(developerLlm.some(entry => JSON.stringify(entry.content).includes(REASONING_TEXT))).toBe(true);
});
it("makes hidden continuity available in agent state before awaited message_end delivery finishes", async () => {
const releaseExtension = Promise.withResolvers<void>();
const extensionEntered = Promise.withResolvers<void>();
const extensionRuntime = new ExtensionRuntime();
const extension = await loadExtensionFromFactory(
pi => {
pi.on("message_end", async () => {
extensionEntered.resolve();
await releaseExtension.promise;
});
},
tempDir.path(),
new EventBus(),
extensionRuntime,
"delayed-message-end",
);
const extensionRunner = new ExtensionRunner(
[extension],
extensionRuntime,
tempDir.path(),
SessionManager.inMemory(),
new ModelRegistry(authStorage, path.join(tempDir.path(), "extension-models.yml")),
);
const harness = createSession(extensionRunner);
const persisted = Promise.withResolvers<void>();
const previous = harness.sessionManager.onEntryAppended;
harness.sessionManager.onEntryAppended = entry => {
previous?.(entry);
if (entry.type !== "custom_message" || entry.customType !== INTERRUPTED_THINKING_MESSAGE_TYPE) return;
harness.sessionManager.onEntryAppended = previous;
persisted.resolve();
};
const message = thinkingAssistant(harness.model, USER_INTERRUPT_LABEL);
harness.session.agent.emitExternalEvent({ type: "message_start", message });
harness.session.agent.emitExternalEvent({ type: "message_end", message });
await extensionEntered.promise;
expect(harness.session.agent.state.messages.some(isInterruptedThinkingMessage)).toBe(true);
releaseExtension.resolve();
await persisted.promise;
});
it("leaves native thinking on non-user aborts and does not append hidden context", async () => {
const harness = createSession();
await emitAssistantEnd(
@@ -11,8 +11,28 @@
* train is heavyweight. The host-idle close logic is therefore extracted
* as {@link reconcileGuestIdleHostState}; this test drives it directly.
*/
import { describe, expect, it, type Mock, mock } from "bun:test";
import { type GuestIdleReconcilerCtx, reconcileGuestIdleHostState } from "@oh-my-pi/pi-coding-agent/collab/guest";
import { afterAll, afterEach, beforeAll, describe, expect, it, type Mock, mock, vi } from "bun:test";
import {
type GuestIdleReconcilerCtx,
type GuestSnapshotActivityReconcilerCtx,
reconcileGuestIdleHostState,
reconcileGuestSnapshotHostState,
} from "@oh-my-pi/pi-coding-agent/collab/guest";
import { resetSettingsForTest, Settings } from "@oh-my-pi/pi-coding-agent/config/settings";
import { StatusLineComponent } from "@oh-my-pi/pi-coding-agent/modes/components/status-line";
beforeAll(async () => {
resetSettingsForTest();
await Settings.init({ inMemory: true });
});
afterAll(() => {
resetSettingsForTest();
});
afterEach(() => {
vi.restoreAllMocks();
});
interface Fixture {
ctx: GuestIdleReconcilerCtx;
@@ -30,6 +50,36 @@ function makeCtx(hasLoader: boolean): Fixture {
return { ctx, markActivityEnd, loaderStop };
}
function makeSession(): ConstructorParameters<typeof StatusLineComponent>[0] {
return {
state: { messages: [], model: undefined },
messages: [],
systemPrompt: [],
agent: { state: { tools: [] } },
skills: [],
isStreaming: false,
isAutoThinking: false,
autoResolvedThinkingLevel: () => undefined,
isFastModeActive: () => false,
isFastModeEnabled: () => false,
getGoalModeState: () => null,
getAsyncJobSnapshot: () => ({ running: [] }),
modelRegistry: { isUsingOAuth: () => false },
sessionFile: "/tmp/collab-guest-idle.jsonl",
sessionManager: {
getSessionName: () => "collab guest idle test",
getUsageStatistics: () => ({
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
premiumRequests: 0,
cost: 0,
}),
},
} as unknown as ConstructorParameters<typeof StatusLineComponent>[0];
}
describe("reconcileGuestIdleHostState", () => {
it("closes the active-time window and stops the loader when the host reports idle", () => {
const { ctx, markActivityEnd, loaderStop } = makeCtx(true);
@@ -67,3 +117,23 @@ describe("reconcileGuestIdleHostState", () => {
expect(loaderStop).toHaveBeenCalledTimes(1);
});
});
describe("reconcileGuestSnapshotHostState", () => {
it("stops the active meter when an idle welcome snapshot finalizes after reconnect", () => {
const statusLine = new StatusLineComponent(makeSession());
let now = 10_000_000;
vi.spyOn(Date, "now").mockImplementation(() => now);
statusLine.markActivityStart();
now += 5_000;
expect(statusLine.getActiveMs()).toBe(5_000);
const ctx: GuestSnapshotActivityReconcilerCtx = {
statusLine,
loadingAnimation: undefined,
};
reconcileGuestSnapshotHostState(ctx, false);
const stoppedAt = statusLine.getActiveMs();
now += 60_000;
expect(statusLine.getActiveMs()).toBe(stoppedAt);
});
});
@@ -175,6 +175,8 @@ function makeGuestContext(counts: number[]): InteractiveModeContext {
setCollabStatus: () => {},
invalidate: () => {},
resetActiveTime: () => {},
markActivityStart: () => {},
markActivityEnd: () => {},
},
ui: { requestRender: () => {} },
chatContainer: { clear: () => {} },
@@ -62,6 +62,37 @@ describe("extension/hook loader process.exit guard (#3680)", () => {
expect(result.errors[0].error).toContain("process.exit(42)");
});
it("converts hard exits from extension and hook factories into load errors", async () => {
const extension = writeModule("factory-exit-extension.ts", "export default function(pi) { process.exit(31); }\n");
const hook = writeModule("factory-exit-hook.ts", "export default function(pi) { process.exit(32); }\n");
const reallyExitExtension = writeModule(
"factory-really-exit-extension.ts",
"export default function(pi) { process.reallyExit(33); }\n",
);
const cwd = project!.path();
const originalExit = process.exit;
const originalReallyExit = process.reallyExit;
const extensionResult = await loadExtensions([extension], cwd);
const hookResult = await loadHooks([hook], cwd);
const reallyExitResult = await loadExtensions([reallyExitExtension], cwd);
expect(process.exit).toBe(originalExit);
expect(process.reallyExit).toBe(originalReallyExit);
expect(extensionResult.extensions).toEqual([]);
expect(extensionResult.errors).toHaveLength(1);
expect(extensionResult.errors[0].path).toBe(extension);
expect(extensionResult.errors[0].error).toContain("process.exit(31)");
expect(hookResult.hooks).toEqual([]);
expect(hookResult.errors).toHaveLength(1);
expect(hookResult.errors[0].path).toBe(hook);
expect(hookResult.errors[0].error).toContain("process.exit(32)");
expect(reallyExitResult.extensions).toEqual([]);
expect(reallyExitResult.errors).toHaveLength(1);
expect(reallyExitResult.errors[0].path).toBe(reallyExitExtension);
expect(reallyExitResult.errors[0].error).toContain("process.reallyExit(33)");
});
it("loads sibling modules even when one of them tries to exit", async () => {
const bad = writeModule("rogue-extension.ts", "process.exit(0)\n");
const good = writeModule(
@@ -27,10 +27,12 @@ import { TempDir } from "@oh-my-pi/pi-utils";
type StubEditor = {
setText: (text: string) => void;
getText: () => string;
clearDraft: (historyText?: string) => void;
addToHistory: Mock<(...args: unknown[]) => unknown>;
onSubmit?: (text: string) => Promise<void>;
pendingImages: ImageContent[];
pendingImageLinks: (string | undefined)[];
imageLinks?: (string | undefined)[];
};
type PromptCustomMessage = Mock<
@@ -65,6 +67,13 @@ function createStubInputControllerContext(opts: {
getText() {
return editorText;
},
clearDraft(historyText?: string) {
if (historyText !== undefined) this.addToHistory(historyText);
this.setText("");
this.imageLinks = undefined;
this.pendingImages = [];
this.pendingImageLinks = [];
},
addToHistory: vi.fn(),
pendingImages: [] as ImageContent[],
pendingImageLinks: [] as (string | undefined)[],
@@ -215,6 +224,34 @@ describe("InputController skill queue chip metadata", () => {
});
expect(promptCustomMessage.mock.calls[0]?.[0].details.__queueChipText).toBeUndefined();
});
it("routes pending images through immediate skill submit and clears the draft", async () => {
const image: ImageContent = { type: "image", data: "aGVsbG8=", mimeType: "image/png" };
const { ctx, editor, promptCustomMessage } = createStubInputControllerContext({
skillCommands,
isStreaming: false,
});
const controller = new InputController(ctx);
controller.setupEditorSubmitHandler();
editor.setText("/skill:test-skill inspect this");
editor.pendingImages = [image];
editor.pendingImageLinks = ["file:///tmp/skill-image.png"];
editor.imageLinks = editor.pendingImageLinks;
await editor.onSubmit?.("/skill:test-skill inspect this");
expect(promptCustomMessage).toHaveBeenCalledTimes(1);
const message = promptCustomMessage.mock.calls[0]?.[0];
if (!message || !Array.isArray(message.content)) {
throw new Error("expected skill prompt to include image content blocks");
}
expect(message.content[0]).toMatchObject({ type: "text", text: expect.stringContaining("Do the thing.") });
expect(message.content[1]).toEqual(image);
expect(editor.getText()).toBe("");
expect(editor.pendingImages).toEqual([]);
expect(editor.pendingImageLinks).toEqual([]);
expect(editor.imageLinks).toBeUndefined();
});
});
describe("compaction skill re-invocation", () => {
@@ -583,6 +620,13 @@ function createStubInteractiveModeContextForUiHelpers(session: AgentSession) {
getText() {
return editorText;
},
clearDraft(historyText?: string) {
if (historyText !== undefined) this.addToHistory(historyText);
this.setText("");
this.imageLinks = undefined;
this.pendingImages = [];
this.pendingImageLinks = [];
},
addToHistory: vi.fn(),
pendingImages: [] as ImageContent[],
pendingImageLinks: [] as (string | undefined)[],
+73 -1
View File
@@ -324,7 +324,9 @@ describe("YieldTool", () => {
},
}),
);
const dataSchema = getSuccessDataSchema(tool.parameters as unknown as Record<string, unknown>);
const dataUnion = getSuccessDataSchema(tool.parameters as unknown as Record<string, unknown>);
// `data` is now a section-variant union; the full-output object is the first branch.
const dataSchema = toRecord(Array.isArray(dataUnion.anyOf) ? dataUnion.anyOf[0] : dataUnion);
const resultsSchema = toRecord(toRecord(dataSchema.properties).results);
const issueSchema = toRecord(toRecord(toRecord(resultsSchema.items).properties).issue);
@@ -340,6 +342,76 @@ describe("YieldTool", () => {
tool.execute("call-mixed-invalid", { result: { data: { results: [{ issue: "185" }] } } } as never),
).rejects.toThrow("Output does not match schema");
});
it("expands section variants so a strict reviewer can submit one incremental section", () => {
const tool = new YieldTool(
createSession({
outputSchema: {
properties: {
overall_correctness: { enum: ["correct", "incorrect"] },
explanation: { type: "string" },
confidence: { type: "number" },
},
optionalProperties: {
findings: {
elements: {
properties: {
title: { type: "string" },
body: { type: "string" },
priority: { type: "number" },
},
},
},
},
},
}),
);
expect(tool.strict).toBe(true);
const toolDefinition: Tool = {
name: tool.name,
description: tool.description,
parameters: tool.parameters,
strict: tool.strict,
};
// One incremental finding (a single element, not the full output) must validate.
expect(
validateToolArguments(toolDefinition, {
type: "toolCall",
id: "call-one-finding",
name: tool.name,
arguments: { type: ["findings"], result: { data: { title: "t", body: "b", priority: 1 } } },
}),
).toBeDefined();
// A lone verdict value must validate too.
expect(
validateToolArguments(toolDefinition, {
type: "toolCall",
id: "call-verdict",
name: tool.name,
arguments: { type: ["overall_correctness"], result: { data: "incorrect" } },
}),
).toBeDefined();
// The full terminal output still validates.
expect(
validateToolArguments(toolDefinition, {
type: "toolCall",
id: "call-full",
name: tool.name,
arguments: {
result: { data: { overall_correctness: "incorrect", explanation: "x", confidence: 0.5 } },
},
}),
).toBeDefined();
// Stays Codex-valid: strict, no top-level combinator.
const [converted] = convertOpenAICodexResponsesTools([toolDefinition], makeCodexModel());
if (converted.type !== "function") throw new Error("expected a function tool payload");
expect(converted.strict).toBe(true);
for (const combinator of ["allOf", "anyOf", "oneOf", "enum", "const", "not"]) {
expect(converted.parameters[combinator]).toBeUndefined();
}
});
it("supports $defs/$ref output schemas by inlining definitions and degrades after first runtime failure", async () => {
const outputSchema = {
$defs: {