Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
19 kB · 460 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461import { createHash } from "node:crypto";import { readFileSync } from "node:fs";import { resolve } from "node:path";import { describe, expect, it } from "vitest";import { isDaemonSessionSummary } from "../src/cli/daemon-launch.js";import { createDaemonCommandEnvelope, createDaemonEventEnvelope, createDaemonEventMeta, createDaemonReplayInfo, DAEMON_COMMAND_COMPATIBILITY, DAEMON_COMMAND_PLANE, DAEMON_DEFAULT_SERVER_CAPABILITIES, DAEMON_OUTBOUND_COMPATIBILITY, DAEMON_PROTOCOL_INFO, DAEMON_PROTOCOL_VERSION, DAEMON_SCHEMA_ID, DAEMON_SCHEMA_REVISION, type DaemonCommand, type DaemonOutbound, getDaemonCommandCompatibilities, isDaemonCommandEnvelope, isDaemonMutatingCommand, isSessionPlaneDaemonCommand, salvageDaemonCommandId,} from "../src/modes/daemon/daemon-protocol.js";import { type DaemonWorkerDescriptor, durableDaemonWorkerDescriptor,} from "../src/modes/daemon/daemon-worker-protocol.js";
describe("daemon protocol helpers", () => { it("serializes worker descriptors as identity-only version 2 state", () => { const descriptor = { version: 1, workerId: "worker", pid: 123, processStartId: "process-start", socketPath: "/tmp/worker.sock", recoveryJournalPath: "/state/recovery.jsonl", orphanProcessJournalPath: "/state/orphans.jsonl", supervisorSocketPath: "/tmp/supervisor.sock", authenticationToken: "local-worker-token", workerInstanceId: "instance-1", rootActiveSessionId: "active", sessionFile: "/sessions/root.jsonl", createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z", lifecycle: "ready", createCommand: { type: "create", sessionPath: "/sessions/root.jsonl", config: { sessionDir: "/legacy/sessions", telemetryDisabled: true, apiKey: "secret-api-key", extensionFlagValues: { providerSecretKey: "secret-extension" }, }, env: { PROVIDER_TOKEN: "secret-client-env" }, launchEnv: { PROVIDER_TOKEN: "secret-launch-env" }, runtimeMetadata: { parentActiveSessionId: "secret-runtime" }, }, launchEnv: { PROVIDER_TOKEN: "secret-top-level-env" }, consecutiveFailures: 0, lastError: "secret-error", } as unknown as DaemonWorkerDescriptor;
const durable = durableDaemonWorkerDescriptor(descriptor);
expect(durable.version).toBe(2); expect(durable.createCommand).toEqual({ type: "create", sessionPath: "/sessions/root.jsonl" }); expect(durable).toMatchObject({ workerId: "worker", workerInstanceId: "instance-1", sessionFile: "/sessions/root.jsonl", sessionDir: "/legacy/sessions", telemetryDisabled: true, }); expect(JSON.stringify(durable)).not.toContain("secret-"); });
it("keeps the advertised schema identity synchronized with wire type shapes", () => { const source = readFileSync(resolve(__dirname, "../src/modes/daemon/daemon-protocol.ts"), "utf8"); const commandSource = source.slice( source.indexOf("export type DaemonCommand ="), source.indexOf("type DaemonCommandName"), ); const savedSessionSource = source.slice( source.indexOf("export interface DaemonSavedSessionInfo"), source.indexOf("export type DaemonDeleteSavedSessionResult"), ); const outboundSource = source.slice( source.indexOf("export type DaemonOutbound ="), source.indexOf("export const DAEMON_OUTBOUND_COMPATIBILITY"), ); const digest = createHash("sha256") .update(`${commandSource}\n${savedSessionSource}\n${outboundSource}`) .digest("hex") .slice(0, 12); expect(DAEMON_SCHEMA_ID).toBe(`protocol-${DAEMON_PROTOCOL_VERSION}-schema-${DAEMON_SCHEMA_REVISION}-${digest}`); });
it("requires compatibility metadata for the heartbeat protocol surface", () => { expect(DAEMON_PROTOCOL_VERSION).toBe(7); expect(DAEMON_SCHEMA_ID).toContain(`protocol-${DAEMON_PROTOCOL_VERSION}`); expect(DAEMON_COMMAND_COMPATIBILITY.heartbeats_list).toEqual({ minProtocol: 7, capability: "heartbeat_catalog", }); expect(DAEMON_COMMAND_COMPATIBILITY.heartbeat_manage).toEqual({ minProtocol: 7, capability: "heartbeat_management", }); expect(DAEMON_COMMAND_COMPATIBILITY.complete_owned_session).toEqual({ minProtocol: 7, capability: "client_owned_sessions", }); expect(DAEMON_OUTBOUND_COMPATIBILITY.heartbeats_changed).toEqual({ minProtocol: 7, capability: "heartbeat_catalog", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toEqual( expect.arrayContaining(["heartbeat_catalog", "heartbeat_management"]), ); });
it("capability-gates explicit subagent deletion instead of schema-gating it", () => { expect(DAEMON_COMMAND_COMPATIBILITY.delete_rlm_subagent).toEqual({ minProtocol: 7, capability: "delete_rlm_subagent", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("delete_rlm_subagent"); });
it("capability- and schema-gates ACP MCP server replacement", () => { expect(DAEMON_COMMAND_COMPATIBILITY.replace_acp_mcp_servers).toEqual({ minProtocol: 7, minSchemaRevision: 22, capability: "acp_mcp_servers", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("acp_mcp_servers"); });
it("capability-gates the optional model catalog surface", () => { expect(DAEMON_COMMAND_COMPATIBILITY.get_model_catalog).toEqual({ minProtocol: 7, capability: "model_catalog", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("model_catalog"); });
it("capability- and schema-gates queued message mutation at its introducing revision", () => { expect(DAEMON_COMMAND_COMPATIBILITY.mutate_queued_message).toEqual({ minProtocol: 7, minSchemaRevision: 15, capability: "queue_message_mutation", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("queue_message_mutation"); });
it("schema-gates the RLM max depth commands at their introducing revision", () => { expect(DAEMON_COMMAND_COMPATIBILITY.get_rlm_max_depth_status).toEqual({ minProtocol: 7, minSchemaRevision: 11 }); expect(DAEMON_COMMAND_COMPATIBILITY.set_rlm_max_depth).toEqual({ minProtocol: 7, minSchemaRevision: 11 }); });
it("schema-gates session commands that carry the telemetry policy", () => { expect(getDaemonCommandCompatibilities({ type: "create", config: { cwd: "/tmp" } })).toEqual([ { minProtocol: 7 }, ]); expect( getDaemonCommandCompatibilities({ type: "create", config: { cwd: "/tmp", telemetryDisabled: true } }), ).toEqual([{ minProtocol: 7, minSchemaRevision: 14 }, { minProtocol: 7 }]); expect(getDaemonCommandCompatibilities({ type: "attach", activeSessionId: "active-1" })).toEqual([ { minProtocol: 7 }, ]); expect( getDaemonCommandCompatibilities({ type: "attach", activeSessionId: "active-1", telemetryDisabled: true }), ).toEqual([{ minProtocol: 7, minSchemaRevision: 14 }, { minProtocol: 7 }]); expect( getDaemonCommandCompatibilities({ type: "reattach", activeSessionId: "active-1", targetActiveSessionId: "active-2", telemetryDisabled: true, }), ).toEqual([{ minProtocol: 7, minSchemaRevision: 14 }, { minProtocol: 7 }]); });
it("capability-gates authoritative rosters and transient owned-session recovery context", () => { expect(DAEMON_COMMAND_COMPATIBILITY.get_rlm_children).toEqual({ minProtocol: 7, minSchemaRevision: 17, capability: "authoritative_child_roster", }); expect( getDaemonCommandCompatibilities({ type: "attach", activeSessionId: "active-1", recoveryConfig: { cwd: "/tmp/fresh-owner" }, }), ).toEqual([ { minProtocol: 7, minSchemaRevision: 17, capability: "owned_session_recovery_context" }, { minProtocol: 7 }, ]); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toEqual( expect.arrayContaining([ "authoritative_child_roster", "owned_session_recovery_context", "rlm_quiescence_barrier", ]), ); });
it("gates the opt-in RLM quiescence wire field", () => { expect( getDaemonCommandCompatibilities({ type: "wait_for_headless_completion", activeSessionId: "active-1", waitForRlmQuiescence: true, }), ).toEqual([{ minProtocol: 7, minSchemaRevision: 18, capability: "rlm_quiescence_barrier" }, { minProtocol: 7 }]); expect( getDaemonCommandCompatibilities({ type: "wait_for_headless_completion", activeSessionId: "active-1", }), ).toEqual([{ minProtocol: 7 }]); });
it("capability- and schema-gates session input pause leases", () => { expect(DAEMON_COMMAND_COMPATIBILITY.acquire_session_input_pause).toEqual({ minProtocol: 7, minSchemaRevision: 19, capability: "session_input_pause", }); expect(DAEMON_COMMAND_COMPATIBILITY.release_session_input_pause).toEqual( DAEMON_COMMAND_COMPATIBILITY.acquire_session_input_pause, ); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("session_input_pause"); });
it("version- and capability-gates prompt admission cancellation", () => { expect(DAEMON_COMMAND_COMPATIBILITY.cancel_prompt_admission).toEqual({ minProtocol: 7, minSchemaRevision: 8, capability: "prompt_admission_cancellation", }); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("prompt_admission_cancellation"); });
it("capability-gates cancellation after prompt ownership", () => { const legacy = { type: "cancel_prompt_admission", activeSessionId: "active-1", admissionId: "a-1" } as const; expect(getDaemonCommandCompatibilities(legacy)).toEqual([DAEMON_COMMAND_COMPATIBILITY.cancel_prompt_admission]); expect(getDaemonCommandCompatibilities({ ...legacy, cancelOwned: true })).toEqual([ { minProtocol: 7, minSchemaRevision: 20, capability: "owned_prompt_cancellation" }, DAEMON_COMMAND_COMPATIBILITY.cancel_prompt_admission, ]); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("owned_prompt_cancellation"); });
it("gates honest worker-state reporting at its introducing schema revision", () => { // Revision 16 adds the "stopping" workerState and stops reporting // disconnected workers as "ready". The field is optional and old clients // ignore unknown values, so no capability gate is needed; the revision // lets version probes distinguish daemons with the old semantics. expect(DAEMON_SCHEMA_REVISION).toBeGreaterThanOrEqual(16); });
it("keeps refine failure events backward-compatible on the existing session event channel", () => { const event: DaemonOutbound = { type: "session_event", activeSessionId: "active-1", event: { type: "refine_failed", error: "disk full" }, };
// Refine events remain on the original session-event channel across later schema revisions. expect(DAEMON_SCHEMA_REVISION).toBeGreaterThanOrEqual(6); expect(DAEMON_OUTBOUND_COMPATIBILITY.session_event).toEqual({ minProtocol: 7 }); expect(event).toMatchObject({ event: { type: "refine_failed", error: "disk full" } }); });
it("accepts legacy side-question and bash shapes in new daemons and clients", () => { const oldClientSideQuestion: DaemonCommand = { type: "start_side_question", activeSessionId: "active-1", sideQuestionId: "side-1", question: "What changed?", }; const oldClientBash: DaemonCommand = { type: "execute_bash", activeSessionId: "active-1", command: "ls", }; const oldDaemonBashStart: DaemonOutbound = { type: "session_event", activeSessionId: "active-1", event: { type: "bash_start", command: "ls", excludeFromContext: false }, }; const oldDaemonBashEnd: DaemonOutbound = { type: "session_event", activeSessionId: "active-1", event: { type: "bash_end", exitCode: 0, cancelled: false, truncated: false }, };
expect(DAEMON_COMMAND_COMPATIBILITY.start_side_question).toEqual({ minProtocol: 7 }); expect(DAEMON_COMMAND_COMPATIBILITY.execute_bash).toEqual({ minProtocol: 7 }); expect(DAEMON_OUTBOUND_COMPATIBILITY.session_event).toEqual({ minProtocol: 7 }); expect(oldClientSideQuestion).not.toHaveProperty("previousTurns"); expect(oldClientBash).not.toHaveProperty("transient"); expect(oldClientBash).not.toHaveProperty("runId"); expect(oldDaemonBashStart.event).not.toHaveProperty("transient"); expect(oldDaemonBashStart.event).not.toHaveProperty("runId"); expect(oldDaemonBashEnd.event).not.toHaveProperty("transient"); expect(oldDaemonBashEnd.event).not.toHaveProperty("runId"); expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toEqual( expect.arrayContaining(["side_question_transcript", "transient_bash"]), ); });
it("creates versioned command and event envelopes", () => { const command = { id: "cmd-1", type: "attach", activeSessionId: "active-1" } as const; const commandEnvelope = createDaemonCommandEnvelope(command, "cmd-1", "client-1"); const eventMeta = createDaemonEventMeta("active-1", 3, "2026-01-01T00:00:00.000Z"); const event: DaemonOutbound = { type: "session_event", activeSessionId: "active-1", event: { type: "agent_end", messages: [] }, meta: eventMeta, };
expect(commandEnvelope).toEqual({ type: "command", id: "cmd-1", protocol: DAEMON_PROTOCOL_INFO, clientId: "client-1", command, }); expect(createDaemonEventEnvelope(event, eventMeta)).toEqual({ type: "event", id: "active-1:3", protocol: DAEMON_PROTOCOL_INFO, activeSessionId: "active-1", sequence: 3, cursor: { generation: "active-1", sequence: 3 }, emittedAt: "2026-01-01T00:00:00.000Z", event, }); expect(eventMeta.cursor).toEqual({ generation: "active-1", sequence: 3 }); });
it("rejects command envelopes from pre-session-action protocols", () => { const command = { id: "cmd-1", type: "attach", activeSessionId: "active-1" } as const;
expect(isDaemonCommandEnvelope(createDaemonCommandEnvelope(command, "cmd-1", "client-1", 7))).toBe(true); expect(isDaemonCommandEnvelope(createDaemonCommandEnvelope(command, "cmd-1", "client-1", 6))).toBe(false); });
it("keeps attachment routing out of the durable mutation journal", () => { expect(isDaemonMutatingCommand({ type: "attach" })).toBe(false); expect(isDaemonMutatingCommand({ type: "reattach" })).toBe(false); // Journal replays after a reconnect would skip re-subscribing the new socket. expect(isDaemonMutatingCommand({ type: "roster_subscribe" })).toBe(false); expect(isDaemonMutatingCommand({ type: "roster_unsubscribe" })).toBe(false); expect(isDaemonMutatingCommand({ type: "wait_for_headless_completion" })).toBe(true); expect(isDaemonMutatingCommand({ type: "switch_session" })).toBe(true); });
it("keeps the roster push additive for pre-roster clients", () => { // Subscription commands and the push are capability-gated; a client that // never sends roster_subscribe is never written a roster_update. expect(DAEMON_COMMAND_COMPATIBILITY.roster_subscribe).toEqual({ minProtocol: 7, capability: "agent_roster" }); expect(DAEMON_COMMAND_COMPATIBILITY.roster_unsubscribe).toEqual({ minProtocol: 7, capability: "agent_roster" }); expect(DAEMON_OUTBOUND_COMPATIBILITY.roster_update).toEqual({ minProtocol: 7, capability: "agent_roster" }); // list responses now carry rosterStatus/statusLabel/lastHeardFromAt; the // summary validator pre-roster clients shipped stays open to additive fields. expect( isDaemonSessionSummary({ id: "session-1", activeSessionId: "active-1", rosterStatus: "running", statusLabel: "queued", lastHeardFromAt: "2026-08-01T12:00:00.000Z", }), ).toBe(true); });
it("capability-gates direct worker transport discovery as a supervisor-only surface", () => { expect(DAEMON_COMMAND_COMPATIBILITY.get_direct_worker_transport).toEqual({ minProtocol: 7, minSchemaRevision: 25, capability: "direct_peer_transport", }); // Only the supervisor issues tickets; workers and standalone daemons must not advertise it. expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).not.toContain("direct_peer_transport"); expect(isDaemonMutatingCommand({ type: "get_direct_worker_transport" })).toBe(false); });
it("classifies every command plane and never defaults unknown commands to the session plane", () => { // A worker "list" means only that worker's sessions; the supervisor list is authoritative. expect(DAEMON_COMMAND_PLANE.list).toBe("control"); expect(DAEMON_COMMAND_PLANE.prompt).toBe("session"); expect(isSessionPlaneDaemonCommand("no_such_command")).toBe(false); });
it("reports replay availability from resume cursors", () => { expect(createDaemonReplayInfo(undefined, 5, "generation-1")).toEqual({ status: "complete", toSequence: 5, toCursor: { generation: "generation-1", sequence: 5 }, }); expect( createDaemonReplayInfo( { activeSessionId: "active-1", generation: "generation-1", sequence: 5 }, 5, "generation-1", ), ).toEqual({ status: "complete", fromSequence: 5, toSequence: 5, fromCursor: { generation: "generation-1", sequence: 5 }, toCursor: { generation: "generation-1", sequence: 5 }, }); expect(createDaemonReplayInfo({ generation: "generation-1", sequence: 10 }, 5, "generation-1")).toEqual({ status: "unavailable", fromSequence: 10, toSequence: 5, fromCursor: { generation: "generation-1", sequence: 10 }, toCursor: { generation: "generation-1", sequence: 5 }, reason: "resume_cursor_ahead_of_session", }); expect(createDaemonReplayInfo({ generation: "generation-1", sequence: 2 }, 5, "generation-1")).toEqual({ status: "unavailable", fromSequence: 2, toSequence: 5, fromCursor: { generation: "generation-1", sequence: 2 }, toCursor: { generation: "generation-1", sequence: 5 }, reason: "event_replay_not_available", }); expect(createDaemonReplayInfo({ generation: "old", sequence: 5 }, 0, "new")).toMatchObject({ status: "unavailable", reason: "event_generation_changed", fromCursor: { generation: "old", sequence: 5 }, toCursor: { generation: "new", sequence: 0 }, }); });
it("salvages command ids from rejected lines regardless of shape validity", () => { const oldEnvelope = JSON.stringify( createDaemonCommandEnvelope({ type: "list" } as DaemonCommand, "list-1", "old-client", 6), ); expect(salvageDaemonCommandId(oldEnvelope)).toBe("list-1"); expect(salvageDaemonCommandId(JSON.stringify({ type: "list", id: "bare-1" }))).toBe("bare-1"); expect(salvageDaemonCommandId(JSON.stringify({ type: null, id: "typeless-1" }))).toBe("typeless-1"); expect(salvageDaemonCommandId(JSON.stringify({ id: "no-type" }))).toBe("no-type"); expect(salvageDaemonCommandId(JSON.stringify({ type: "command", id: 7 }))).toBeUndefined(); expect(salvageDaemonCommandId(JSON.stringify("command"))).toBeUndefined(); expect(salvageDaemonCommandId("{ not json")).toBeUndefined(); });});