Skip to content
Merged
9 changes: 7 additions & 2 deletions CONTEXT.md
Original file line number Diff line number Diff line change
Expand Up @@ -192,7 +192,9 @@ with `volli session answer <handle>` naming the rest), so the child's prose
never reads as the parent's user (VC-457). The parent is never parked on it,
nothing bounds the child's time, and stopping the parent stops its children. A
child a person resumes after that notice reports to its parent again when the
resumed turn ends. The Session
resumed turn ends; a turn a host notice opens in it (its own shell exiting, a
Browser Tab event) does not, unless a person or another Session steers a
message into that turn (VC-809). The Session
Presentation Contract projects that notice into the parent's transcript as a
quiet host-authored receipt row, not as a Turn in the person's voice. This row
records the delegation outcome; it is not a listing of the child. While work is
Expand Down Expand Up @@ -475,7 +477,10 @@ SIGKILL. The three names ride one port, appended last to the vocabulary. A
shell belongs to the Session that started it — a subagent never sees its
parent's — is capped per Session, runs inside the workspace, is spawned
through the same environment record and the same attachment identity the
`execute` tool gets, and is killed when the attachment ends. Every shell
`execute` tool gets, and is killed when the attachment ends — a subagent's
already when a turn of it completes, silently, so a shell it left running
cannot reopen it after its parent has the answer (VC-809); its exited shells'
output is forgotten at that point too. Every shell
result restates the Session's live shells, because the tool calls are the
durable record; shells are live resources, not ledger facts, and do not
survive a relaunch. The Activity Island's shells cluster reads them through
Expand Down
5 changes: 3 additions & 2 deletions packages/agent-runtime/src/pi/shell-tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -273,15 +273,16 @@ const DESCRIPTIONS: Record<ShellToolId, string> = {
"Start a command that runs beside the turn instead of holding it: a dev server, a watch build, a long test run, a log tail.",
"Returns the shell's id and whatever it printed in its first second, so a server's listening line comes back in the same call.",
"When it exits, Volli sends you a notice with how it ended and the end of its output, by itself: do not poll or sleep for it. Optional notifyOn adds one notice the first time output matches, even before a newline.",
"Volli caps how many a Session may hold and kills every one when the Session's attachment ends. Use execute for a command you want to wait on.",
"Volli caps how many a Session may hold and kills every one when the Session's attachment ends. A Subagent Session's shells end silently when its turn completes, so a subagent reads what it needs or waits inside the turn.",
"Use execute for a command you want to wait on.",
].join(" "),
shell_output: [
"Read a background shell's output. By default this returns only what is new since the last read of that shell, so polling costs the same each time;",
"pass tail to get the last N bytes of everything retained instead. The result states whether the shell is still running or has exited, and with what.",
].join(" "),
shell_kill: [
"End a background shell: SIGTERM, then SIGKILL if it lingers.",
"Its output stays readable with shell_output until the Session's attachment ends.",
"Its output stays readable with shell_output until the Session's attachment ends, or for a Subagent Session until its turn completes.",
].join(" "),
};

Expand Down
2 changes: 2 additions & 0 deletions packages/host-core/src/agent-tool-door.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1461,6 +1461,7 @@ describe("session_delegate through the Agent Tool Surface (VC-9)", () => {
delegate: () => ({
watching: () => false,
rearm: async () => undefined,
rearmJoined: async () => undefined,
delegate: async (input) => {
delegated.push(input);
if (input.task.includes("overflow")) {
Expand Down Expand Up @@ -1725,6 +1726,7 @@ describe("verb result details match the registry (VC-471)", () => {
delegate: () => ({
watching: () => false,
rearm: async () => undefined,
rearmJoined: async () => undefined,
delegate: async (input) => ({
childSessionId: CHILD,
handle: CHILD.slice(0, 8),
Expand Down
28 changes: 19 additions & 9 deletions packages/host-core/src/session-runtime/agents.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import { actorSessionTicketDisplay } from "../agent-dispatch/resolution";
import type { AgentToolDoor } from "../agent-tool-door";
import { createHostAgentToolDoor, createHostAgentWatches } from "../agent-services";
import type { Watches } from "../watches";
import { createDelegations, type Delegations } from "./delegate-session";
import { commandOriginOf, createDelegations, type Delegations } from "./delegate-session";
import type { TicketSessionDelegationStore } from "./delegation-store";
import type { RuntimeSessionFacade, RuntimeSessionServices } from "./facade";
import { readRecoveredSessionServices, type RecoveredSessionServices } from "./lifecycle";
Expand Down Expand Up @@ -73,17 +73,27 @@ export function createRuntimeSessionAgents(options: RuntimeSessionAgentOptions)
const store = sessionDelegation;
unsubscribeDelegations = sessionWakeBus.subscribe((wake) => {
const payload = wake.event.payload;
if (payload.kind !== "turn.started" || created.watching(wake.event.sessionId)) return;
// A turn's start re-arms a settled child; so does a message steered
// into a turn already running (VC-809), which that start never saw.
const joined =
payload.kind === "command.recorded" && payload.command.intent.kind === "message.submit";
if (payload.kind !== "turn.started" && !joined) return;
if (created.watching(wake.event.sessionId)) return;
const entry = store.subagentDelegation(wake.event.sessionId);
if (entry === null) return;
const work = created
.rearm(entry, { turnId: payload.turnId, afterSequence: wake.event.sequence })
.catch((error: unknown) => {
options.log.error("could not re-arm the notice for a resumed subagent", {
childSessionId: entry.childSessionId,
error,
});
const work = (
payload.kind === "turn.started"
? created.rearm(entry, { turnId: payload.turnId, afterSequence: wake.event.sequence })
: created.rearmJoined(entry, {
origin: commandOriginOf(wake.event),
afterSequence: wake.event.sequence,
})
).catch((error: unknown) => {
options.log.error("could not re-arm the notice for a resumed subagent", {
childSessionId: entry.childSessionId,
error,
});
});
liveHost?.detachedWork.track(work);
});
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,13 @@ afterEach(async () => {
const QUESTION = "Which implementation should continue?";
const OPTIONS = [{ id: "first", label: "First implementation" }];
const provenance = { source: { kind: "system" as const, id: "test", detail: null }, venue: null };
// The runtime's queue admission bound covers EVERY attach (its workspace and
// executor stages) and every cold binding rebuild, not only queued messages.
// A short wall-clock bound on all of them made a merely slow real Pi attach
// under CI load a false `rejected` (VC-778). Genuine work keeps the production
// bound; only a stage this fixture deliberately stalls gets the short one.
const PRODUCTION_ADMISSION_BOUND_MS = 30_000;
const STALLED_STAGE_BOUND_MS = 50;

function fixture() {
const database = openTestDb();
Expand Down Expand Up @@ -63,7 +70,17 @@ function fixture() {
attachFailure: Error | null;
attachGate: Promise<void> | null;
admissionGate: Promise<void> | null;
} = { attachFailure: null, attachGate: null, admissionGate: null };
/** Real delay before every attach, to model a slow but healthy executor. */
attachDelayMs: number;
/** Set synchronously on entering a gate, before the runtime arms its timer. */
stalled: boolean;
} = {
attachFailure: null,
attachGate: null,
admissionGate: null,
attachDelayMs: 0,
stalled: false,
};
const adapter = createPiNativeAdapter({
sessionDataDir,
models: script.models,
Expand All @@ -90,13 +107,20 @@ function fixture() {
...adapter,
attach: async (spec, sink) => {
attachSpecs.push(spec);
if (control.attachGate) await control.attachGate;
if (control.attachGate) {
control.stalled = true;
await control.attachGate;
}
if (control.attachDelayMs > 0)
await new Promise((resolve) => setTimeout(resolve, control.attachDelayMs));
if (control.attachFailure) throw control.attachFailure;
const binding = await adapter.attach(spec, sink);
const dispatch = binding.dispatch.bind(binding);
binding.dispatch = async (command) => {
if (command.kind === "message.submit" && control.admissionGate)
if (command.kind === "message.submit" && control.admissionGate) {
control.stalled = true;
await control.admissionGate;
}
return dispatch(command);
};
return binding;
Expand All @@ -105,7 +129,11 @@ function fixture() {
followUps: createSqliteSessionFollowUpLedger(database.db),
// Explicit recovery wakes exercise retry; no background backoff races.
followUpRetryDelaysMs: [],
followUpAdmissionTimeoutMs: 50,
// Read when each bounded stage arms its timer, which is after the
// wrappers above have entered (and flagged) any gate synchronously.
get followUpAdmissionTimeoutMs() {
return control.stalled ? STALLED_STAGE_BOUND_MS : PRODUCTION_ADMISSION_BOUND_MS;
},
onFollowUpFailure: (error) => {
failures.push(error);
},
Expand Down Expand Up @@ -194,7 +222,7 @@ async function start(f: Host, settledPrefix = false) {
commandId: `attach-${clock++}`,
command: { kind: "adapter.attach", continuity: "fresh" },
});
expect(attached.receipt?.status).toBe("accepted");
expect(attached.receipt?.status, JSON.stringify(attached.receipt)).toBe("accepted");
if (settledPrefix) {
const settled = await f.host.command({
sessionId,
Expand All @@ -208,7 +236,7 @@ async function start(f: Host, settledPrefix = false) {
},
},
});
expect(settled.receipt?.status).toBe("accepted");
expect(settled.receipt?.status, JSON.stringify(settled.receipt)).toBe("accepted");
}
const commandId = `submit-${clock++}`;
const result = await f.host.command({
Expand All @@ -224,7 +252,7 @@ async function start(f: Host, settledPrefix = false) {
},
},
});
expect(result.receipt?.status).toBe("accepted");
expect(result.receipt?.status, JSON.stringify(result.receipt)).toBe("accepted");
await expect
.poll(
async () => (await f.host.projection({ sessionId })).projection.interactions.active.length,
Expand Down Expand Up @@ -254,7 +282,7 @@ async function queue(current: Launch, sessionId: string, commandId = "queue-next
},
},
});
expect(result.receipt?.status).toBe("accepted");
expect(result.receipt?.status, JSON.stringify(result.receipt)).toBe("accepted");
await current.host.recoverFollowUps();
}

Expand Down Expand Up @@ -340,6 +368,24 @@ async function assertNoAnswers(current: Launch, sessionId: string) {
// The provider alone is scripted. Session admission, SQLite events/queue,
// Pi sidecars, boot recovery and the adapter's parked ask are production code.
describe("crash recovery through the durable Pi binding (VC-753)", () => {
it("accepts a slow but healthy attach and recovers its binding without a false admission timeout (VC-778)", async () => {
const f = fixture();
// Longer than the stalled-stage bound: a fixture that bounded genuine
// work by it rejected the fresh attach and failed the boot rebuild.
const slow = STALLED_STAGE_BOUND_MS * 3;
f.control.attachDelayMs = slow;
const { sessionId } = await start(f);
const current = await f.restart(f, true);
current.control.attachDelayMs = slow;
expect(await f.boot(current)).toEqual({ closed: 0, errors: [] });
expect(current.attachSpecs.map(({ continuity }) => continuity)).toEqual(["native_resume"]);
const recovered = (await current.host.projection({ sessionId })).projection;
expect(recovered.liveExecutor).not.toBeNull();
expect(
recovered.attention.active.filter(({ kind }) => kind === "adapter_unrecoverable"),
).toEqual([]);
});

it("reconciles an accepted command with a different persisted timestamp without a false recovery failure", async () => {
const f = fixture();
const { sessionId, commandId } = await start(f);
Expand Down Expand Up @@ -655,7 +701,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => {
sessionId,
command: { kind: "adapter.release", attachmentId },
});
expect(released.receipt?.status).toBe("accepted");
expect(released.receipt?.status, JSON.stringify(released.receipt)).toBe("accepted");
current.control.attachFailure = new Error("Scripted attach unavailable");
await queue(current, sessionId);
const projection = (await current.host.projection({ sessionId })).projection;
Expand All @@ -678,7 +724,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => {
sessionId,
command: { kind: "adapter.attach", continuity: "context_replay" },
});
expect(attached.receipt?.status).toBe("accepted");
expect(attached.receipt?.status, JSON.stringify(attached.receipt)).toBe("accepted");
await assertDeliveredOnce(f, current, sessionId);
});

Expand Down Expand Up @@ -751,7 +797,10 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => {
expect.objectContaining({
id: expect.stringContaining("message-delivery:"),
kind: "adapter_unrecoverable",
detail: expect.stringContaining("queue_admission_timeout"),
// The stalled stage, not a slow real rebuild before it, is what timed out.
detail: expect.stringContaining(
`queue_admission_timeout: ${stage === "attach" ? "executor attach" : "message admission"} did not finish`,
),
}),
);
expect(f.script.requests).toHaveLength(2);
Expand All @@ -761,6 +810,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => {
gate.reject(new Error("Stalled fixture transport ended"));
current.control.attachGate = null;
current.control.admissionGate = null;
current.control.stalled = false;
}
const relaunched = await f.restart(current);
const projection = (await relaunched.host.projection({ sessionId })).projection;
Expand Down
Loading
Loading