diff --git a/apps/p2/docs/p2-notes.md b/apps/p2/docs/p2-notes.md index 7bd5ee4ef..521e2992c 100644 --- a/apps/p2/docs/p2-notes.md +++ b/apps/p2/docs/p2-notes.md @@ -444,7 +444,9 @@ navigates to its view before awaiting the worker, so a failed first start also h Host `onError`, refused reloads and worker failures publish bounded, redacted owner notices (`LogNotices`), with Show logs and Dismiss. Notices keep the latest error per source (at most 50 sources); normal shutdown/abort messages do not become worker-failure notices. Archived-session admission refusals are not cached as crashes, -so restoring an archive still works without an extra Retry. Neither guests nor workers may read or dismiss notices, or invoke +so restoring an archive still works without an extra Retry. Archiving stops the worker with +`SessionWorkerControl.suspend`, not `stop`: holders of the session's objects (a page's lingering conversation) wait +suspended until a restore (`retry`) or their release, instead of starting a worker the archive refuses. Neither guests nor workers may read or dismiss notices, or invoke the worker retry control. Dev controls additionally publish their own reload/restart results. The owner-only `logs` plugin adds **Logs** to `sidebar-foot:logs`. Its sheet selects Server, session workers, @@ -543,7 +545,7 @@ translated phone drawer and intercept chat taps. Chord v2's `buildFacet` against the API. - **Runtime registries and locals** (`src/runtime`): `Routes` (HTTP by `METHOD /path`, exact path), `WorkerInitValues` (spawn parameters contributed per plugin; a throw refuses the start, e.g. archived sessions), `Hosts` (worker - states, stop), `HarnessHost`/`HarnessNative`, `Extensions`, `HarnessConfig`, `Slots`, + states, restart for the holders), `HarnessHost`/`HarnessNative`, `Extensions`, `HarnessConfig`, `Slots`, `useObject`/`useReadable`/`useStatus`. - **No configuration process globals.** Diagnostics alone have native process-scoped buffers because they capture stdout/stderr before/after object activation and survive facet reload. Host configuration is the core's: `Server` (server: `info` = data dir, diff --git a/apps/p2/plugins/search/server.ts b/apps/p2/plugins/search/server.ts index 6d3566aee..714bf6201 100644 --- a/apps/p2/plugins/search/server.ts +++ b/apps/p2/plugins/search/server.ts @@ -137,7 +137,7 @@ export default defineFacet((host) => { } finally { if (archived) { await archivedStarts.end({ session }, ctx); - await hosts.stop({ session }, ctx); + await server().workers.suspend(session); } } }; diff --git a/apps/p2/plugins/sessions/SessionOptions.tsx b/apps/p2/plugins/sessions/SessionOptions.tsx index da3d13060..7a31c4001 100644 --- a/apps/p2/plugins/sessions/SessionOptions.tsx +++ b/apps/p2/plugins/sessions/SessionOptions.tsx @@ -1,11 +1,12 @@ // Native catalog operations beneath the copied mobile Session sheet. -import { BACKGROUND_CONTEXT } from "@p2/app/v1"; +import { BACKGROUND_CONTEXT, liveIn, whenLive } from "@p2/app/v1"; import { closeLayers, conversationRoute, MoreIcon, openLayer, useComposerState, + useHost, useNavigation, useObject, useReadable, @@ -16,7 +17,8 @@ import { directoryPickerLayer } from "../envs/contract.web.tsx"; import { Sessions } from "./contract.ts"; import { SessionSheet } from "./SessionSheet.tsx"; export function SessionOptions() { - const nav = useNavigation(), + const host = useHost(), + nav = useNavigation(), route = useRoute(); const service = useObject(Sessions, {}, { variant: "owner" }), rows = useReadable(service?.sessions); @@ -47,7 +49,14 @@ export function SessionOptions() { onArchive={() => { closeLayers(); void nav.replace(conversationRoute(route, undefined)); - perform(service?.archive({ id: session.id }, BACKGROUND_CONTEXT)); + // Leaving the session unmounts this view and releases `service`, which cancels its calls: the + // archive holds a reference of its own until it is done. + const archiving = host.hold(Sessions, {}, BACKGROUND_CONTEXT); + perform( + whenLive(archiving) + .then(() => liveIn(archiving, "owner")?.archive({ id: session.id }, BACKGROUND_CONTEXT)) + .finally(() => archiving.release()), + ); }} /> )} diff --git a/apps/p2/plugins/sessions/server.ts b/apps/p2/plugins/sessions/server.ts index 4b6f813ba..c99e22a2a 100644 --- a/apps/p2/plugins/sessions/server.ts +++ b/apps/p2/plugins/sessions/server.ts @@ -18,7 +18,7 @@ import { serveKeyed, WorkerBoundary, } from "@p2/app/v1"; -import { Hosts, Server, serverOf, WorkerInitValues } from "@p2/app/v1/server"; +import { Server, serverOf, WorkerInitValues } from "@p2/app/v1/server"; import { EnvManagers, EnvService } from "../envs/contract.ts"; import { ModelConfig, ModelResources } from "../models/contract.ts"; import { ChildStore } from "../subagents/contract.ts"; @@ -236,7 +236,6 @@ export default defineFacet((host) => { return { variant: "guest", members: { sessions: sessionsView } }; } if (!isOwner(principal)) throw forbidden("sessions are owner-only"); - const hosts = self.get(Hosts, {}); const lifecycleOf = serverOf(self); const models = self.get(ModelConfig, {}); const resources = nativeOf(self, ModelResources, "The model configuration"); @@ -361,7 +360,9 @@ export default defineFacet((host) => { } catch (error) { if (created) { await catalog.update({ id: copy, archived: true }, ctx).catch(() => {}); - await hosts.stop({ session: copy }, ctx).catch(() => {}); + await lifecycleOf() + .workers.suspend(copy) + .catch(() => {}); } else await removeSessionCopy(dataDir, copy); throw error; } @@ -376,10 +377,13 @@ export default defineFacet((host) => { await catalog.update({ id, pinned }, ctx); return null; }, - archive: async ({ id }, ctx) => { - await catalog.update({ id, archived: true }, ctx); - // Archive stops the worker whatever it is doing; its storage and history stay. - await hosts.stop({ session: id }, ctx); + archive: async ({ id }) => { + await lifecycle(async (root) => { + await catalog.update({ id, archived: true }, root); + // Archive stops the worker whatever it is doing; its storage and history stay. Its holders (a page's + // lingering conversation) wait for a restore rather than start a worker the archive refuses. + await lifecycleOf().workers.suspend(id); + }); return null; }, restore: async ({ id }, ctx) => { diff --git a/apps/p2/src/hosts/session-workers.ts b/apps/p2/src/hosts/session-workers.ts index c6ce3f244..adc315811 100644 --- a/apps/p2/src/hosts/session-workers.ts +++ b/apps/p2/src/hosts/session-workers.ts @@ -233,6 +233,20 @@ export class SessionWorkers implements SessionWorkerControl { ); } + /** + * Stops an archived session's worker for good: its holders (a page's lingering conversation, say) wait suspended + * instead of starting a worker the archived session refuses, which failed with an owner notice. `retry` (restore) + * ends the suspension; so does the release of its last holder (a later hold is refused as archived). + */ + async suspend(session: string): Promise { + this.#crashRestarts.delete(session); + await this.#serially(session, async () => { + await this.workers.stop(Session({ id: session }), BACKGROUND_CONTEXT); + for (const notice of processLogs.notices()) + if (notice.source === `${SESSION_PREFIX}${session}`) processLogs.dismissNotice(notice.id); + }); + } + /** Stops every worker; their exits are no news to subscribers (the server's objects close next). */ close(context: Context): Promise { this.#listeners.clear(); diff --git a/apps/p2/src/runtime/server.ts b/apps/p2/src/runtime/server.ts index bb94f669b..1e274e054 100644 --- a/apps/p2/src/runtime/server.ts +++ b/apps/p2/src/runtime/server.ts @@ -85,7 +85,13 @@ export interface SessionWorkerControl { ): Promise; /** Owner-requested retry: stops the previous instance and clears the failure before a fresh reference. */ retry(session: string): Promise; + /** Stops the session's worker; its holders start a new one (a restart, e.g. for a changed environment). */ stop(session: string): Promise; + /** + * Stops an archived session's worker and starts none for its holders: they wait (suspended) until `retry` (restore) + * or their release, instead of a start the archived session refuses. + */ + suspend(session: string): Promise; /** Every change of a worker's state. */ subscribe(listener: (event: HostEvent) => void): () => void; } @@ -234,7 +240,7 @@ export const WorkerStateSchema = obj({ export type WorkerState = Static; export type HostEvent = { readonly session: string } & WorkerState; -/** Session worker processes: which run (by session ID), and stopping one (archive). */ +/** Session worker processes: which run (by session ID), and restarting one for its holders (`stop`). */ export const Hosts = local("p2.hosts", { version: 2, params: {}, diff --git a/apps/p2/test/p2/archive.test.ts b/apps/p2/test/p2/archive.test.ts new file mode 100644 index 000000000..85a651f6d --- /dev/null +++ b/apps/p2/test/p2/archive.test.ts @@ -0,0 +1,74 @@ +// Archiving a session stops its worker for good: holders of its objects (a page's lingering conversation) wait for a +// restore instead of starting a worker the archived session refuses (a failed start with an owner notice). + +import assert from "node:assert/strict"; +import { test } from "node:test"; +import type { AssistantMessage } from "@earendil-works/pi-ai"; +import type { ConversationView } from "@earendil-works/pi-durable"; +import { BACKGROUND_CONTEXT, whenLive } from "@p2/app/v1"; +import { Chat } from "../../plugins/chat/contract.ts"; +import { Sessions } from "../../plugins/sessions/contract.ts"; +import { SessionStatus } from "../../plugins/status/contract.ts"; +import { processLogs } from "../../src/runtime/logs.ts"; +import { failureOf, holdIn, settled, startFixture, until } from "./helpers.ts"; + +const lastAnswer = (view: ConversationView | null | undefined): string => { + const message = view?.entries.at(-1)?.model?.[0] as AssistantMessage | undefined; + return message?.role === "assistant" + ? message.content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("") + : ""; +}; + +test( + "an archived session's held objects wait for a restore; no worker starts meanwhile", + { timeout: 90_000 }, + async (t) => { + const fixture = await startFixture({ + plugins: [ + "chat", + "conversations", + "drafts", + "envs", + "files", + "models", + "search", + "sessions", + "status", + "subagents", + ], + }); + t.after(() => fixture.close()); + const browser = fixture.browser(); + const sessions = await holdIn(browser, Sessions, {}, "owner"); + const status = await holdIn(browser, SessionStatus, {}, "reader"); + const { id } = await sessions.create({ cwd: fixture.workspace }, BACKGROUND_CONTEXT); + const source = `session:${id}`; + const starts = () => processLogs.lines(source).filter((line) => line.text.startsWith("Starting worker")).length; + // A page's conversation, still held (lingering) when the session is archived. + const chat = await holdIn(browser, Chat, { session: id, conversation: 1 }, "participant"); + await chat.send({ text: "hello", requestId: "r1" }, BACKGROUND_CONTEXT); + await until(chat.view, (view) => lastAnswer(view) === "Echo: hello"); + assert.equal(starts(), 1); + + await sessions.archive({ id }, BACKGROUND_CONTEXT); + await until(status.statuses, (value) => value?.[id]?.running === false); + const failure = await failureOf(chat); + assert.match(failure.message, /suspended/); + // Nothing started a worker for the held conversation, so nothing failed as archived. + await new Promise((resolve) => setTimeout(resolve, 1000)); + assert.equal(starts(), 1); + assert.equal((await settled(status.statuses))[id]?.error, undefined); + assert.deepEqual( + processLogs.notices().filter((notice) => notice.source === source), + [], + ); + + // Restoring starts a worker for the waiting holder. + await sessions.restore({ id }, BACKGROUND_CONTEXT); + await whenLive(chat); + await chat.send({ text: "restored", requestId: "r2" }, BACKGROUND_CONTEXT); + await until(chat.view, (view) => lastAnswer(view) === "Echo: restored"); + assert.equal(starts(), 2); + chat.release(); + }, +); diff --git a/apps/p2/test/p2/helpers.ts b/apps/p2/test/p2/helpers.ts index 3dec49d79..bea3a13c1 100644 --- a/apps/p2/test/p2/helpers.ts +++ b/apps/p2/test/p2/helpers.ts @@ -284,6 +284,7 @@ const NO_WORKERS: SessionWorkerControl = { }, retry: async () => {}, stop: async () => {}, + suspend: async () => {}, subscribe: () => () => {}, }; diff --git a/apps/p2/test/p2/ui-mobile-fixes.test.ts b/apps/p2/test/p2/ui-mobile-fixes.test.ts index 2d6039417..f0a7ee446 100644 --- a/apps/p2/test/p2/ui-mobile-fixes.test.ts +++ b/apps/p2/test/p2/ui-mobile-fixes.test.ts @@ -408,4 +408,29 @@ describe("Pi Mobile audit fixes in the browser", { skip: !browserTests }, () => `location.pathname === '/' && document.querySelector('.toast')?.textContent === 'Session is archived'`, ); }); + + it("archives the viewed session from its sheet without an error or a notice", async () => { + const sessions = await holdIn(owner, Sessions, {}, "owner"); + const session = await sessions.create({ cwd: fixture.workspace, title: "Archived here" }, BACKGROUND_CONTEXT); + await browser.metrics(390, 844); + await browser.open(url(`/s/${session.id}`)); + await browser.wait(`document.querySelector('.composer textarea') && !document.querySelector('.offline')`); + // The page's conversation is live: its worker runs, and the page keeps holding it after leaving (linger). + const hosts = fixture.server.host.hold(Hosts, {}, BACKGROUND_CONTEXT); + await until(hosts.workers, (workers) => workers?.[session.id]?.running === true); + await browser.click('[aria-label="Session options"]'); + await browser.click(".row-button.danger"); + await until(sessions.sessions, (list) => Boolean(list?.find((s) => s.id === session.id)?.archivedAt)); + await until(hosts.workers, (workers) => workers?.[session.id]?.running === false); + // Leaving the session released the sheet's reference: the archive's call is not cancelled with it (no + // "p2.sessions: released" toast), and the lingering conversation starts no worker the archive refuses (no notice). + await new Promise((resolve) => setTimeout(resolve, 1000)); + assert.deepEqual( + await browser.evaluate( + `({ path: location.pathname, toast: document.querySelector('.toast')?.textContent ?? null, notices: [...document.querySelectorAll('.owner-notice')].map((e) => e.textContent) })`, + ), + { path: "/", toast: null, notices: [] }, + ); + hosts.release(); + }); });