Merge pull request #10763 from esengine/maint/v1-issue-fixes

fix(v1): issue fixes for 1.39.x stability and regressions
This commit is contained in:
YHH authored and GitHub committed 2026-09-25 06:03:44 -07:00
commit 27d99bbdb0
88 files changed
+2305 -198

No files matched your search

+102 -3
View File
@@ -6,7 +6,7 @@ import { test } from "node:test";
import type { ServiceState } from "../shared/ipc.js";
import { validateHelloResult, type HelloResult } from "./handshake.js";
import { RestartBudget } from "./restartBudget.js";
import { ServiceSupervisor } from "./service.js";
import { ServiceSupervisor, ShutdownError, type ShutdownTiming } from "./service.js";
const silent = { info() {}, warn() {}, error() {} };
const tick = async (times = 4) => {
@@ -19,9 +19,11 @@ class FakeChild extends EventEmitter {
stderr = new PassThrough();
alive = true;
requests: Array<{ id: number; method: string; params: unknown }> = [];
status: Record<string, unknown> | null = null;
heldShutdown: number | null = null;
private buffered = "";
constructor(readonly generation: string, readonly behaviour: { helloError?: { code: number; message: string }; exitOnStdinEnd?: boolean;
constructor(readonly generation: string, readonly behaviour: { helloError?: { code: number; message: string }; exitOnStdinEnd?: boolean; holdShutdown?: boolean;
shutdownResults?: Array<Record<string, unknown>>; },) {
super();
this.stdin.on("data", (chunk: Buffer) => {
@@ -57,9 +59,24 @@ class FakeChild extends EventEmitter {
this.send({ method: "desktop/event", params: { seq: 1, generation, name, args: [{ ok: true }] }, });
}
replyShutdown(result: Record<string, unknown>): void {
if (this.heldShutdown === null) throw new Error("no shutdown request is held");
this.send({ id: this.heldShutdown, result: { requestId: "", reason: "user_quit", retryable: false, ...result } });
this.heldShutdown = null;
}
private handle(frame: { id?: number; method?: string; params?: unknown }): void {
if (typeof frame.id !== "number" || typeof frame.method !== "string") return;
this.requests.push({ id: frame.id, method: frame.method, params: frame.params, });
if (frame.method === "desktop/shutdown" && this.behaviour.holdShutdown) {
this.heldShutdown = frame.id;
return;
}
if (frame.method === "desktop/shutdownStatus" && this.status) {
const params = frame.params as { requestId?: string };
this.send({ id: frame.id, result: { requestId: params.requestId ?? "", reason: "user_quit", retryable: false, ...this.status } });
return;
}
if (frame.method === "desktop/hello") {
if (this.behaviour.helloError) {
this.send({ id: frame.id, error: this.behaviour.helloError });
@@ -107,7 +124,7 @@ class FakeChild extends EventEmitter {
}
}
function harness(options: { children?: FakeChild[]; budget?: RestartBudget; onState?(state: ServiceState): void; } = {},) {
function harness(options: { children?: FakeChild[]; budget?: RestartBudget; onState?(state: ServiceState): void; shutdownTiming?: Partial<ShutdownTiming>; } = {},) {
const spawned: FakeChild[] = [];
const states: ServiceState[] = [];
const events: string[] = [];
@@ -123,6 +140,7 @@ function harness(options: { children?: FakeChild[]; budget?: RestartBudget; onSt
log: silent,
budget: options.budget ?? new RestartBudget(),
exitGraceMs: 10,
shutdownTiming: options.shutdownTiming,
spawn: () => {
const child = options.children?.[index++] ?? new FakeChild(`g-${spawned.length + 1}`, {});
spawned.push(child);
@@ -290,6 +308,87 @@ test("shutdown status polling waits through in-progress cleanup", async () => {
);
});
const inProgress = (phase: string) => ({ phase, outcome: "in_progress", completed: false });
const completed = { phase: "completed", outcome: "success", completed: true };
const fastShutdown: ShutdownTiming = { replyMs: 40, pollMs: 5, stallMs: 150, ceilingMs: 2_000 };
const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
test("a shutdown reply that outlives the reply window is still a success when it arrives", async () => {
const child = new FakeChild("g-1", { holdShutdown: true });
child.status = inProgress("saving");
const h = harness({ children: [child], shutdownTiming: fastShutdown });
await h.supervisor.start();
const phases: string[] = [];
const shutdown = h.supervisor.shutdown("user_quit", (phase) => phases.push(phase));
await wait(60);
child.status = inProgress("closing");
await wait(80);
child.replyShutdown(completed);
child.exit(0, null);
await shutdown;
assert.equal(h.supervisor.current.phase, "exited");
assert.equal(h.failures.length, 0);
assert.ok(phases.includes("closing"), `phases: ${phases.join(",")}`);
assert.equal(phases.at(-1), "completed");
});
test("a late completed status ends a shutdown whose reply never arrives", async () => {
const child = new FakeChild("g-1", { holdShutdown: true });
child.status = inProgress("saving");
const h = harness({ children: [child], shutdownTiming: fastShutdown });
await h.supervisor.start();
const shutdown = h.supervisor.shutdown();
await wait(70);
child.status = inProgress("closing");
await wait(70);
child.status = completed;
await shutdown;
assert.equal(child.alive, false);
assert.equal(h.supervisor.current.phase, "exited");
});
test("a shutdown that stops advancing is reported incomplete and the service is kept", async () => {
const child = new FakeChild("g-1", { holdShutdown: true });
child.status = inProgress("closing");
const h = harness({ children: [child], shutdownTiming: fastShutdown });
await h.supervisor.start();
const started = Date.now();
await assert.rejects(h.supervisor.shutdown(), (error: unknown) => {
assert.ok(error instanceof ShutdownError);
assert.equal(error.code, "shutdown_incomplete");
assert.match(error.message, /closing/);
return true;
});
assert.ok(Date.now() - started < fastShutdown.ceilingMs, "a stall is reported before the ceiling");
assert.equal(child.alive, true, "an incomplete shutdown leaves the service for a retry");
});
test("a service that exits without reporting success fails the shutdown", async () => {
const child = new FakeChild("g-1", { holdShutdown: true });
child.status = inProgress("saving");
const h = harness({ children: [child], shutdownTiming: fastShutdown });
await h.supervisor.start();
const shutdown = h.supervisor.shutdown();
await wait(60);
child.exit(1, null);
await assert.rejects(shutdown, (error: unknown) => error instanceof ShutdownError && error.code === "service_exited");
});
test("a shutdown that keeps advancing still ends at the ceiling", async () => {
const child = new FakeChild("g-1", { holdShutdown: true });
const phases = ["saving", "closing"];
let turn = 0;
const flip = setInterval(() => { child.status = inProgress(phases[turn++ % 2]); }, 10);
child.status = inProgress("saving");
const h = harness({ children: [child], shutdownTiming: { ...fastShutdown, ceilingMs: 300 } });
await h.supervisor.start();
try {
await assert.rejects(h.supervisor.shutdown(), (error: unknown) => error instanceof ShutdownError && error.code === "shutdown_incomplete");
} finally {
clearInterval(flip);
}
});
test("shutdown fences a hello completion queued before shutdown", async () => {
const h = harness();
const starting = h.supervisor.start();
+97 -44
View File
@@ -8,9 +8,33 @@ import { RestartBudget } from "./restartBudget.js";
import { RpcClient } from "./rpc.js";
export const LIFECYCLE_TIMEOUT_MS = 10_000;
const SHUTDOWN_STATUS_POLL_MS = 250;
export const EXIT_GRACE_MS = 5_000;
export interface ShutdownTiming {
replyMs: number; // silence on desktop/shutdown before its status is followed
pollMs: number;
stallMs: number; // longest a live shutdown may keep one phase and outcome
ceilingMs: number; // hard bound on the whole wait, however it advances
}
export const SHUTDOWN_TIMING: ShutdownTiming = {
replyMs: LIFECYCLE_TIMEOUT_MS,
pollMs: 250,
stallMs: 30_000,
ceilingMs: 60_000,
};
// code is the failure's identity; the message is only its display form.
export class ShutdownError extends Error {
constructor(
readonly code: string,
detail: string,
) {
super(`${code}: ${detail}`);
this.name = "ShutdownError";
}
}
export interface ServiceHandlers {
hello(client: RpcClient): Promise<HelloResult>;
onRequest(method: string, params: unknown): Promise<unknown>;
@@ -32,6 +56,7 @@ export interface ServiceOptions {
budget?: RestartBudget;
now?: () => number;
exitGraceMs?: number;
shutdownTiming?: Partial<ShutdownTiming>;
}
interface Session {
@@ -81,6 +106,15 @@ function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function within<T>(promise: Promise<T>, ms: number): Promise<T | null> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([promise, new Promise<null>((resolve) => (timer = setTimeout(() => resolve(null), ms)))]);
} finally {
clearTimeout(timer);
}
}
export class ServiceSupervisor {
private session: Session | null = null;
private state: ServiceState = { phase: "starting", generation: "" };
@@ -95,6 +129,7 @@ export class ServiceSupervisor {
private readonly spawnFn: SpawnFn;
private readonly now: () => number;
private readonly exitGraceMs: number;
private readonly shutdownTiming: ShutdownTiming;
constructor(
private readonly options: ServiceOptions,
@@ -104,6 +139,7 @@ export class ServiceSupervisor {
this.spawnFn = options.spawn ?? ((command, args, spawnOptions) => nodeSpawn(command, args, spawnOptions));
this.now = options.now ?? (() => Date.now());
this.exitGraceMs = options.exitGraceMs ?? EXIT_GRACE_MS;
this.shutdownTiming = { ...SHUTDOWN_TIMING, ...options.shutdownTiming };
}
get current(): ServiceState {
@@ -201,52 +237,12 @@ export class ServiceSupervisor {
}
session.expectExit = true;
if (!this.shutdownRequestId) this.shutdownRequestId = randomUUID();
let result: ShutdownResult | null = null;
try {
result = shutdownResult(
await session.client.request(
"desktop/shutdown",
{
requestId: this.shutdownRequestId,
reason,
},
LIFECYCLE_TIMEOUT_MS,
),
);
} catch (error) {
this.options.log.warn(`desktop/shutdown result unknown: ${errorText(error)}; querying status`);
result = shutdownResult(
await session.client.request(
"desktop/shutdownStatus",
{
requestId: this.shutdownRequestId,
},
LIFECYCLE_TIMEOUT_MS,
),
);
}
const deadline = Date.now() + LIFECYCLE_TIMEOUT_MS;
onProgress?.(result.phase);
while (!result.completed && result.outcome === "in_progress" && Date.now() < deadline) {
await new Promise((resolve) => setTimeout(resolve, SHUTDOWN_STATUS_POLL_MS));
result = shutdownResult(
await session.client.request(
"desktop/shutdownStatus",
{
requestId: this.shutdownRequestId,
},
LIFECYCLE_TIMEOUT_MS,
),
);
onProgress?.(result.phase);
}
const result = await this.awaitShutdown(session, reason, onProgress);
if (result.outcome === "failed") {
throw new Error(`${result.errorCode ?? "shutdown_failed"}: ${result.error ?? "desktop shutdown failed"}`);
throw new ShutdownError(result.errorCode ?? "shutdown_failed", result.error ?? "desktop shutdown failed");
}
if (!result.completed || result.outcome !== "success") {
throw new Error(
`${result.errorCode ?? "shutdown_incomplete"}: ${result.error ?? `desktop shutdown stopped in ${result.phase}`}`,
);
throw new ShutdownError(result.errorCode ?? "shutdown_incomplete", result.error ?? `desktop shutdown stopped in ${result.phase}`);
}
// The service closes itself only after the completed result has reached the
// shell. stdin/kill are now a post-completion process-exit fallback.
@@ -254,6 +250,63 @@ export class ServiceSupervisor {
this.setState({ phase: "exited", generation: "" });
}
// The shutdown reply stays awaited for the whole wait: a completed result that
// arrives after replyMs is the service's verdict, not an orphan. Status polls
// only decide whether the shutdown is still live and advancing meanwhile.
private async awaitShutdown(
session: Session,
reason: "user_quit" | "update_restart" | "system_signal",
onProgress?: (phase: ShutdownPhase) => void,
): Promise<ShutdownResult> {
const timing = this.shutdownTiming;
const requestId = this.shutdownRequestId;
const startedAt = this.now();
const reply = session.client
.request("desktop/shutdown", { requestId, reason }, timing.ceilingMs)
.then((value) => ({ result: shutdownResult(value) }))
.catch((error: unknown) => ({ error }));
const first = await within(reply, timing.replyMs);
const unanswered = new Promise<never>(() => undefined);
const answer = first ? unanswered : reply.then((settled) => ("result" in settled ? settled.result : unanswered));
let result = null as ShutdownResult | null;
let advancedAt = startedAt;
const observe = (next: ShutdownResult) => {
if (!result || next.phase !== result.phase || next.outcome !== result.outcome) {
advancedAt = this.now();
onProgress?.(next.phase);
}
result = next;
};
if (first && "result" in first) observe(first.result);
else {
const why = first ? errorText(first.error) : `no reply after ${timing.replyMs} ms`;
this.options.log.warn(`desktop/shutdown result unknown: ${why}; following status`);
}
for (;;) {
if (result && (result.completed || result.outcome !== "in_progress")) return result;
const phase = result?.phase ?? "unknown";
if (this.now() - startedAt >= timing.ceilingMs) {
throw new ShutdownError("shutdown_incomplete", `desktop shutdown did not complete within ${timing.ceilingMs} ms; last phase ${phase}`);
}
if (this.now() - advancedAt >= timing.stallMs) {
throw new ShutdownError("shutdown_incomplete", `desktop shutdown stopped in ${phase}`);
}
const late = await within(answer, timing.pollMs);
if (late) {
observe(late);
continue;
}
if (!session.alive) {
throw new ShutdownError("service_exited", `desktop service exited during shutdown in phase ${phase}`);
}
try {
observe(shutdownResult(await session.client.request("desktop/shutdownStatus", { requestId }, timing.replyMs)));
} catch (error) {
if (session.alive) this.options.log.warn(`desktop/shutdownStatus unanswered: ${errorText(error)}`);
}
}
}
private live(): Session {
const session = this.session;
if (this.stopping) throw new Error("desktop service is shutting down");
@@ -0,0 +1,43 @@
import assert from "node:assert/strict";
import { JSDOM } from "jsdom";
import type { WireAsk } from "../lib/types";
const dom = new JSDOM('<div id="root"></div>', { url: "http://localhost/", pretendToBeVisual: true });
Object.assign(globalThis, {
window: dom.window, document: dom.window.document, Element: dom.window.Element,
HTMLElement: dom.window.HTMLElement, Node: dom.window.Node, localStorage: dom.window.localStorage,
IS_REACT_ACT_ENVIRONMENT: true,
});
Object.defineProperty(globalThis, "navigator", { configurable: true, value: dom.window.navigator });
const { default: React, act } = await import("react");
const { createRoot } = await import("react-dom/client");
const { AskCard } = await import("../components/AskCard");
const { LocaleProvider } = await import("../lib/i18n");
const root = createRoot(document.getElementById("root")!);
const ask: WireAsk = {
id: "1", runtimeEpoch: "runtime-a", turnId: "turn-a",
questions: [{ id: "q1", prompt: "Choose", options: [{ label: "Option A" }, { label: "Option B" }] }],
};
await act(async () => root.render(<LocaleProvider><AskCard ask={ask} draftScope="tab-a" onAnswer={() => {}} onStop={() => {}} /></LocaleProvider>));
const card = () => document.querySelector<HTMLElement>(".prompt-shelf__card")!;
const collapsed = () => card().classList.contains("prompt-shelf__card--collapsed");
const click = async (element: Element) => {
await act(async () => { element.dispatchEvent(new dom.window.MouseEvent("click", { bubbles: true })); });
};
assert.equal(collapsed(), false, "a delivered Ask starts expanded");
await click(document.querySelector(".ask-shelf__custom-indicator")!);
assert.equal(collapsed(), false, "clicking the custom-answer row outside its input keeps the card open");
assert.equal(document.activeElement, document.querySelector(".ask-shelf__custom"), "the row click focuses the custom input");
await click(document.querySelector(".ask-shelf__custom-row")!);
assert.equal(collapsed(), false, "clicking the row's own padding keeps the card open");
console.log(" PASS the custom-answer row never toggles the card's collapse");
await click(document.querySelector(".prompt-shelf__heading")!);
assert.equal(collapsed(), true, "the card's own chrome still collapses it");
console.log(" PASS non-interactive card chrome still collapses the Ask");
await act(async () => root.unmount());
dom.window.close();
@@ -169,7 +169,10 @@ assert.equal([...document.querySelectorAll<HTMLButtonElement>('.composer-workspa
const graphAction = [...document.querySelectorAll('.composer-workspace-menu--branches [role="menuitem"]')].find((item) => item.textContent?.includes("Git graph"));
await click(graphAction ?? null);
assert.equal(calls.history, 1);
assert(document.querySelector('[role="dialog"][aria-labelledby="composer-git-graph-title"]'));
const graphDialog = document.querySelector('[role="dialog"][aria-labelledby="composer-git-graph-title"]');
assert(graphDialog);
assert.ok(graphDialog.closest(".modal-backdrop")?.parentElement === document.body, "the Git graph escapes the composer frame's isolated stacking context");
assert.ok(!rootElement.contains(graphDialog), "the Git graph is not painted inside the composer tree");
assert(document.body.textContent?.includes("Ship workspace context"));
assert(document.body.textContent?.includes("abcdef1"));
await click(document.querySelector('.composer-git-graph__actions button[aria-label="Close"]'));
@@ -18,6 +18,7 @@ let polledPreparation: SessionPreparationView | undefined;
let navigationEpoch = 0;
let cancelled = 0;
const opened: string[] = [];
const preparedSelectors: unknown[] = [];
const source = { hostId: "local", sourceKey: "legacy", path: "/fixture/legacy.jsonl" };
const host = installDesktopHostStub({
CheckHistoricalSourceUpdate: async () => ({ sourceKey: "legacy", status: "available", version: "v2", source, retryable: false }),
@@ -30,6 +31,11 @@ const host = installDesktopHostStub({
if (polledPreparation) return polledPreparation;
throw new Error("terminal preparation must not poll");
},
PrepareSession: async (selector: unknown) => {
preparedSelectors.push(selector);
return { operationId: "prepare-source", sourceKey: "legacy", status: "ready", revision: 8,
target: { hostId: "local", sessionId: "imported-source" }, retryable: false };
},
CancelSessionPreparation: async (operationId: string) => { cancelled++; return cancellation ? cancellation.promise : { operationId, sourceKey: "legacy", status: "cancelled", revision: 3, retryable: true }; },
});
const root = createRoot(document.getElementById("root")!);
@@ -117,10 +123,12 @@ await act(async () => root.render(<LocaleProvider><HistoricalSessionBanners
tab={{ ...baseProps.tab, sessionId: undefined, ready: false, historicalSource: source }}
navigate={async intent => { pendingIntents.push(intent); }}
/></LocaleProvider>));
assert.deepEqual(pendingIntents, [], "restoring a legacy tab never starts preparation automatically");
assert.deepEqual(preparedSelectors, [], "restoring a legacy tab never starts preparation automatically");
assert.ok(document.body.textContent?.includes("Import this conversation to keep sending messages"), "the banner states why sending waits");
await act(async () => document.getElementById("reasonix-prepare-restored-session")!.click());
assert.equal(pendingIntents.length, 1);
assert.equal((pendingIntents[0] as { kind: string }).kind, "resume-session", "explicit preparation uses the shared navigation owner");
assert.deepEqual(preparedSelectors, [{ source }], "the action prepares the source instead of reopening it in place");
assert.deepEqual(pendingIntents, [{ kind: "canonical-session", ref: { hostId: "local", sessionId: "imported-source" } }],
"a prepared source opens as its canonical session");
const [canonicalTab] = seedActiveTabMetaList([{ ...baseProps.tab, historicalSource: source }], baseProps.tab);
assert.equal(canonicalTab.historicalSource, undefined, "canonical metadata omission clears the old preparation state");
await act(async () => root.render(<LocaleProvider><HistoricalSessionBanners {...baseProps} tab={canonicalTab} /></LocaleProvider>));
@@ -76,6 +76,12 @@ function Probe() {
const root = createRoot(document.getElementById("root")!);
try {
await act(async () => { root.render(<Probe />); });
await act(async () => { await commands.handleOpenTopic("project", "/workspace", "existing-topic", "session-id:existing-session"); });
assert.deepEqual(enqueued.at(-1)?.request, {
kind: "canonical-session", ref: { hostId: "local", sessionId: "existing-session" },
}, "sidebar session keeps its exact local identity on the canonical session path");
enqueued.length = 0;
intent = 0;
await act(async () => {
await commands.openBlankSession("global", "/ignored/global/root");
await commands.openBlankSession("project", "/workspace");
@@ -5,7 +5,7 @@ import { createRoot } from "react-dom/client";
import { JSDOM } from "jsdom";
import { initialState } from "../lib/useController";
import { applyHydrateErrorState } from "../lib/hydrateErrorState";
import { projectSessionAvailability, type SessionAvailability } from "../lib/sessionAvailability";
import { projectSessionAvailability, submitBlockReason, type SessionAvailability } from "../lib/sessionAvailability";
assert.deepEqual(projectSessionAvailability({empty:true}),{kind:"ready",source:"history"},"an empty welcome has no runtime to wait for");
assert.equal(projectSessionAvailability({}).kind,"loading","an unresolved selected source still reports loading");
@@ -26,6 +26,13 @@ const failedLocal = applyHydrateErrorState(readyLocal, "startup", "history read
assert.deepEqual(projectSessionAvailability({ local: { ...failedLocal, meta: {
...readyLocal.meta, ready: false, historicalSource: { hostId: "local", path: "/fixture/legacy.jsonl" },
} } }), { kind: "pending", source: "runtime" }, "unprepared history is a user choice, not a failed recovery");
const nativeHistory = projectSessionAvailability({ local: { ...readyLocal, meta: {
...readyLocal.meta, historicalSource: { hostId: "local", path: "/fixture/legacy.jsonl" },
} } });
assert.deepEqual(nativeHistory, { kind: "pending", source: "runtime" }, "a readable native history runtime still cannot accept sends");
const recordKey = ((key: string) => key) as Translator;
assert.equal(submitBlockReason(nativeHistory, recordKey), "sessionRecovery.importBeforeSend", "the disabled composer says why and what to do");
assert.equal(submitBlockReason(projectSessionAvailability({ local: readyLocal }), recordKey), undefined);
const noop = () => {};
const common: TranscriptSurfaceProjectionInput = {
hydrating: false, hydrateHistoryLoaded: undefined, hydratePlaceholderItems: undefined, hydratePlaceholderActive: false,
@@ -38,6 +38,7 @@ import { buildAppShellClassNames, buildSessionStatusBannerProps, buildSidebarReg
import { buildBottomRegionsProps, buildWorkspaceDockProps } from "./dockRegionBuilders";
import { buildOverlayHostProps } from "./overlayBuilders";
import { buildComposerSurface, buildDecisionFooterSurface, buildFooterTodo, buildFooterUndo } from "./decisionFooterBuilders";
import { submitBlockReason } from "../lib/sessionAvailability";
const WindowsWindowControls = lazy(() => import("./WindowsWindowControls").then((module) => ({ default: module.WindowsWindowControls })));
const DockLauncher = lazy(() => import("../components/DockLauncher").then((module) => ({ default: module.DockLauncher })));
@@ -405,8 +406,7 @@ export function AppRuntimeView(props: AppRuntimeViewProps) {
decisionActive: Boolean(decisionSurface),
runtimeTransitioning: presentationTransitioning,
controllerReady: controllerReady && session.transcript.availability.kind === "ready",
submitDisabledReason: session.transcript.availability.kind !== "ready" && session.transcript.availability.source !== "runtime"
? t("sessionRecovery.sendAfterRecovery") : undefined,
submitDisabledReason: submitBlockReason(session.transcript.availability, t),
showContextWindowRing: false,
},
base: conversationView.composer,
+3 -1
View File
@@ -384,7 +384,9 @@ function AskCardBody({ ask, onAnswer, onStop, draftKey }: AskCardProps & { draft
<div
className={`ask-shelf__custom-row${custom[q.id]?.trim() ? " ask-shelf__custom-row--active" : ""}`}
role="group"
onClick={() => {
onClick={(event) => {
// The card toggles collapse on clicks outside its controls; this row is a control.
event.stopPropagation();
setSelectedIndex(customRowIndex);
setAnswerMode((m) => ({ ...m, [q.id]: "custom" }));
setCustomOpen(true);
@@ -1,5 +1,6 @@
import { ErrorMessage } from "./ErrorMessage";
import { useCallback, useEffect, useMemo, useRef, useState, type RefObject } from "react";
import { createPortal } from "react-dom";
import { Check, ChevronDown, Cloud, Folder, FolderOpen, GitBranch, GitGraph, MessageCircle, Plus, RefreshCw, Search, X } from "lucide-react";
import { asArray } from "../lib/array";
import { app, onProjectTreeChanged } from "../lib/bridge";
@@ -120,7 +121,9 @@ function ComposerGitGraphDialog({
};
}, [load, onClose, restoreFocusRef]);
return (
// The composer frame is an isolated stacking context, so an in-place
// backdrop would paint under the workspace dock.
return createPortal(
<div className="modal-backdrop composer-git-graph-backdrop" data-app-overlay="" onMouseDown={(event) => {
if (event.target === event.currentTarget) onClose();
}}>
@@ -159,7 +162,8 @@ function ComposerGitGraphDialog({
))}
</div>
</section>
</div>
</div>,
document.body,
);
}
@@ -146,7 +146,9 @@ export function HistoricalSessionBanners({ tab, navigate, captureNavigation }: H
const cancellingOperationRef = useRef("");
const activeHostId = activeRef?.hostId ?? "";
const activeSessionId = activeRef?.sessionId ?? "";
const activeKey = activeSessionId ? `${activeHostId}:${activeSessionId}` : "";
const historicalSource = tab?.historicalSource;
const activeKey = activeSessionId ? `${activeHostId}:${activeSessionId}`
: historicalSource ? `source:${tab?.id}:${historicalSource.sourceKey || historicalSource.path}` : "";
const activeKeyRef = useRef(activeKey);
activeKeyRef.current = activeKey;
const mounted = useRef(true);
@@ -174,7 +176,7 @@ export function HistoricalSessionBanners({ tab, navigate, captureNavigation }: H
};
void run().catch(() => {});
return () => { current = false; };
}, [activeHostId, activeSessionId]);
}, [activeHostId, activeSessionId, activeKey]);
const dismissUpdate = () => {
if (update?.version) {
@@ -182,8 +184,17 @@ export function HistoricalSessionBanners({ tab, navigate, captureNavigation }: H
}
setUpdate(null);
};
const importUpdate = async () => {
if (busy || !update?.source || !update.version || !app.PrepareHistoricalSourceVersion || !app.GetSessionPreparation) return;
const importUpdate = () => {
if (!update?.source || !update.version || !app.PrepareHistoricalSourceVersion) return;
const { source, version } = update;
return runPreparation(() => app.PrepareHistoricalSourceVersion!(source, version));
};
const importSource = () => {
if (!historicalSource || !app.PrepareSession) return;
return runPreparation(() => app.PrepareSession!({ source: historicalSource }));
};
const runPreparation = async (start: () => Promise<SessionPreparationView>) => {
if (busy || !app.GetSessionPreparation) return;
const expectedActive = activeKey;
const operation = ++updateOperation.current;
const navigationCurrent = captureNavigation?.() ?? (() => activeKeyRef.current === expectedActive);
@@ -191,7 +202,7 @@ export function HistoricalSessionBanners({ tab, navigate, captureNavigation }: H
setBusy(true);
setUpdateError("");
try {
let view: SessionPreparationView = await app.PrepareHistoricalSourceVersion(update.source, update.version);
let view = await start();
while (current() && !terminalPreparation.has(view.status)) {
await new Promise(resolve => setTimeout(resolve, 300));
if (!current()) return;
@@ -233,14 +244,11 @@ export function HistoricalSessionBanners({ tab, navigate, captureNavigation }: H
{!waiting && preparation.retryable && <button type="button" className="btn btn--small" onClick={() => void navigate({ kind: "resume-session", session: preparation.session })}>{t("common.retry")}</button>}
</div>;
}
if (tab?.historicalSource) return <div className="banner banner--warning banner--actionable" role="status">
<span className="banner__msg">{tab.topicTitle || m("historicalTitle")} · {m("historicalAvailable")}</span>
<span className="banner__hint">{m("historicalImportDescription")}</span>
if (historicalSource) return <div className={`banner ${updateError ? "banner--error" : "banner--warning"} banner--actionable`} role={updateError ? "alert" : "status"}>
<span className="banner__msg">{tab?.topicTitle || m("historicalTitle")} · {m("historicalAvailable")}</span>
<span className="banner__hint">{updateError ? <ErrorMessage error={updateError} /> : m(busy ? "historicalImporting" : "historicalImportToSend")}</span>
<span className="banner__spacer" />
<button id="reasonix-prepare-restored-session" type="button" className="btn btn--small" onClick={() => void navigate({ kind: "resume-session", session: {
source: tab.historicalSource, path: tab.historicalSource!.path, scope: tab.scope, workspaceRoot: tab.workspaceRoot,
topicId: tab.topicId, title: tab.topicTitle, preview: "", turns: 0, turnsState: "unknown", createdAt: 0, lastActivityAt: 0, modTime: 0, current: true, open: true,
} })}>{m("historicalImportOpen")}</button>
<button id="reasonix-prepare-restored-session" type="button" className="btn btn--small" disabled={busy} onClick={() => void importSource()}>{m("historicalImportOpen")}</button>
</div>;
if (!update) return null;
return <div className={`banner ${updateError ? "banner--error" : "banner--warning"} banner--actionable`} role={updateError ? "alert" : "status"}>
@@ -49,6 +49,7 @@ const messages = {
historicalSourceBusy: ["其他实例正在使用此会话,关闭后可重试。", "其他實例正在使用此對話,關閉後可重試。", "In use by another instance. Close it and retry."],
historicalImportFailed: ["导入失败,原件已保留。可重试或查看恢复详情。", "匯入失敗,原件已保留。可重試或檢視恢復詳情。", "Import failed. Original files are preserved; retry or review recovery details."],
historicalAvailable: ["尚未导入", "尚未匯入", "Not imported"],
historicalImportToSend: ["导入后才能继续发送消息,原始文件会保留。", "匯入後才能繼續傳送訊息,原始檔案會保留。", "Import this conversation to keep sending messages. The original file is kept."],
historicalSourceUpdated: ["旧来源有更新,可导入为独立分支", "舊來源有更新,可匯入為獨立分支", "Historical source changed. Import as a separate branch."],
historicalQueued: ["排队中", "排隊中", "Queued"],
historicalImporting: ["正在导入", "正在匯入", "Importing"],
@@ -1,3 +1,4 @@
import type { Translator } from "./i18n";
import type { State } from "./useController";
import type { RemoteSessionApi } from "./useRemoteSession";
@@ -30,3 +31,10 @@ export function projectSessionAvailability(input: { local?: LocalSession; remote
&& (!local.meta.runtime || local.meta.runtime.phase === "ready");
return { kind: ready ? "ready" : "loading", source: "runtime" };
}
/** The reason a disabled composer states, or none when readiness alone explains it. */
export function submitBlockReason(availability: SessionAvailability, t: Translator): string | undefined {
if (availability.kind === "pending") return t("sessionRecovery.importBeforeSend");
if (availability.kind !== "ready" && availability.source !== "runtime") return t("sessionRecovery.sendAfterRecovery");
return undefined;
}
+1
View File
@@ -295,6 +295,7 @@ export const en = {
"sessionRecovery.retryFailed": "Recovery failed. Check the details and try again.",
"sessionRecovery.details": "Details",
"sessionRecovery.sendAfterRecovery": "Send after recovery",
"sessionRecovery.importBeforeSend": "Import to send",
"sessionRecovery.contentAfterRecovery": "Session content will appear after recovery",
"takeover.bannerButton": "Take over session",
"takeover.title": "Take over this session",
+1
View File
@@ -286,6 +286,7 @@ export const zhTW: Record<DictKey, string> = {
"sessionRecovery.retryFailed": "恢復失敗,請查看詳情後重試。",
"sessionRecovery.details": "詳情",
"sessionRecovery.sendAfterRecovery": "恢復後可傳送",
"sessionRecovery.importBeforeSend": "匯入後可傳送",
"sessionRecovery.contentAfterRecovery": "恢復後顯示會話內容",
"takeover.bannerButton": "接管會話",
"takeover.title": "接管該會話",
+1
View File
@@ -296,6 +296,7 @@ export const zh: Record<DictKey, string> = {
"sessionRecovery.retryFailed": "恢复失败,请查看详情后重试。",
"sessionRecovery.details": "详情",
"sessionRecovery.sendAfterRecovery": "恢复后可发送",
"sessionRecovery.importBeforeSend": "导入后可发送",
"sessionRecovery.contentAfterRecovery": "恢复后显示会话内容",
"takeover.bannerButton": "接管会话",
"takeover.title": "接管该会话",
+3
View File
@@ -76,6 +76,9 @@ func (a *App) archiveHistoricalSourceWithOperation(ctx context.Context, id strin
outcome := "archived"
if lifecycle.Lifecycle == workspacestate.Deleted {
outcome = "already_removed"
if err := retireRemovedSourceTopic(state, source); err != nil {
return SessionMutationResult{}, fmt.Errorf("retire removed source topic: %w", err)
}
} else if lifecycle.Lifecycle == workspacestate.Archived {
if _, err := a.desktopSessionService("").Query().Snapshot(ctx, ref); err != nil {
return SessionMutationResult{}, err
@@ -0,0 +1,80 @@
package main
import (
"os"
"path/filepath"
"strings"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/store"
)
// retireRemovedSourceTopic makes clearing a source whose session is already
// deleted durable: its legacy topic joins deletedTopics, which every sidebar
// projection honours. A topic that anything else still carries stays listed.
func retireRemovedSourceTopic(state workspacestate.State, source historicalSource) error {
if source.format == "canonical" {
return nil
}
meta, ok, err := agent.LoadBranchMeta(source.path)
if err != nil || !ok {
return err
}
topicID := strings.TrimSpace(meta.TopicID)
if topicID == "" || topicHasSurvivingOwner(state, topicID) {
return nil
}
shared, err := legacyTopicSharedBeyond(state, source.path, topicID)
if err != nil || shared {
return err
}
return removeTopicFromProjectsFile(topicID)
}
func topicHasSurvivingOwner(state workspacestate.State, topicID string) bool {
for id, presentation := range state.Presentation {
if presentation.TopicID == topicID && state.SessionStates[id].Lifecycle != workspacestate.Deleted {
return true
}
}
for _, pending := range state.PendingCreates {
if pending.Presentation != nil && pending.Presentation.TopicID == topicID {
return true
}
}
return false
}
// legacyTopicSharedBeyond reports whether another transcript beside path
// carries topicID without being adopted by a deleted session. Unreadable
// metadata cannot prove the topic is unshared.
func legacyTopicSharedBeyond(state workspacestate.State, path, topicID string) (bool, error) {
entries, err := os.ReadDir(filepath.Dir(path))
if err != nil {
return false, err
}
for _, entry := range entries {
sibling := filepath.Join(filepath.Dir(path), entry.Name())
if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) || sameDesktopPath(sibling, path) {
continue
}
meta, ok, err := agent.LoadBranchMeta(sibling)
if err != nil || !ok {
return true, nil
}
if strings.TrimSpace(meta.TopicID) == topicID && !sourceAdoptedByDeletedSession(state, sibling) {
return true, nil
}
}
return false, nil
}
func sourceAdoptedByDeletedSession(state workspacestate.State, path string) bool {
for _, mapping := range state.SourceMappings {
if sameDesktopPath(mapping.Path, path) && state.SessionStates[mapping.SessionID].Lifecycle == workspacestate.Deleted {
return true
}
}
return false
}
@@ -0,0 +1,148 @@
package main
import (
"encoding/json"
"os"
"path/filepath"
"slices"
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/session"
)
// Earlier purge writers dropped the deleted session's topic from every receipt,
// so its retained legacy transcript lists again as a live topic. Clearing that
// row must survive a restart (#10735).
func TestAlreadyRemovedSourceStaysRemovedAfterRestart(t *testing.T) {
isolateDesktopUserDirs(t)
path, _, head := migrationSingleDAGFixture(t)
const topicID = "topic_20260903-142302_04afb4407ec2fa2a"
if err := agent.UpdateBranchMeta(path, false, func(m *agent.BranchMeta) error {
m.Scope, m.TopicID, m.TopicTitle = "global", topicID, "Ghost"
return nil
}); err != nil {
t.Fatal(err)
}
if err := saveProjectsFile(desktopProjectFile{GlobalTopics: []string{topicID}}); err != nil {
t.Fatal(err)
}
app := NewApp()
app.ctx = t.Context()
pinDesktopSessionRoot(t, app)
root := app.desktopSessions.root
installNoopRuntimeEvents(app)
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
ref := session.SessionRef{HostID: localDesktopHostID, SessionID: state.SourceMappings[desktopSourceKey(path, head)].SessionID}
if err := app.ArchiveCanonicalSession(ref); err != nil {
t.Fatal(err)
}
if err := app.PurgeCanonicalSession(ref); err != nil {
t.Fatal(err)
}
app.closeSessionServices()
dropTopicEvidence(t, app.workspaceRegistry().Path(), ref.SessionID)
restart := func() *App {
next := NewApp()
next.ctx = t.Context()
next.desktopSessions.root = root
installNoopRuntimeEvents(next)
t.Cleanup(next.closeSessionServices)
installSessionCatalogForTest(t, next, filepath.Dir(path), "global", "")
t.Cleanup(func() { next.stopSessionCatalog(time.Second) })
return next
}
list := func(a *App) []ProjectNode {
t.Helper()
page, err := a.ListProjectTopics(ProjectTopicPageRequest{Scope: "global", Limit: 50})
if err != nil {
t.Fatal(err)
}
a.ReleaseReadSnapshot(page.SnapshotID)
return page.Items
}
app = restart()
rows := list(app)
if len(rows) != 1 || rows[0].TopicID != topicID {
t.Fatalf("fixture must reproduce the residual row: %+v", rows)
}
result, err := app.ArchiveSessionTarget(SessionSelector{Source: &SessionSourceRef{HostID: localDesktopHostID, Path: path, HeadID: head}})
if err != nil || !result.Committed || result.Outcome != "already_removed" {
t.Fatalf("clear residual row = %+v, %v", result, err)
}
app.stopSessionCatalog(time.Second)
app.closeSessionServices()
app = restart()
if rows := list(app); len(rows) != 0 {
t.Fatalf("cleared residual row returned after restart: %+v", rows)
}
}
func dropTopicEvidence(t *testing.T, statePath, sessionID string) {
t.Helper()
body, err := os.ReadFile(statePath)
if err != nil {
t.Fatal(err)
}
var state map[string]any
if err := json.Unmarshal(body, &state); err != nil {
t.Fatal(err)
}
ops, _ := state["pendingOperations"].(map[string]any)
for _, raw := range ops {
op, _ := raw.(map[string]any)
ids, _ := op["sessionIds"].([]any)
if slices.Contains(ids, any(sessionID)) {
delete(op, "presentation")
}
}
if body, err = json.Marshal(state); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(statePath, body, 0o600); err != nil {
t.Fatal(err)
}
if _, err := workspacestate.NewStore(statePath).Load(t.Context()); err != nil {
t.Fatalf("rewritten state: %v", err)
}
}
func TestRemovedSourceTopicStaysWhileAnotherTranscriptCarriesIt(t *testing.T) {
dir := t.TempDir()
const topicID = "topic_shared"
paths := []string{filepath.Join(dir, "removed.jsonl"), filepath.Join(dir, "sibling.jsonl")}
for _, path := range paths {
if err := agent.NewSession("system").Save(path); err != nil {
t.Fatal(err)
}
if err := agent.UpdateBranchMeta(path, false, func(m *agent.BranchMeta) error { m.TopicID = topicID; return nil }); err != nil {
t.Fatal(err)
}
}
state := workspacestate.State{
SessionStates: map[string]workspacestate.SessionState{"gone": {Lifecycle: workspacestate.Deleted}},
SourceMappings: map[string]workspacestate.SourceMapping{"removed": {SourceKey: "removed", Path: paths[0], SessionID: "gone"}},
}
if shared, err := legacyTopicSharedBeyond(state, paths[0], topicID); err != nil || !shared {
t.Fatalf("unadopted sibling transcript = %v, %v; want shared", shared, err)
}
state.SourceMappings["sibling"] = workspacestate.SourceMapping{SourceKey: "sibling", Path: paths[1], SessionID: "gone"}
if shared, err := legacyTopicSharedBeyond(state, paths[0], topicID); err != nil || shared {
t.Fatalf("sibling adopted by a deleted session = %v, %v; want unshared", shared, err)
}
state.Presentation = map[string]workspacestate.Presentation{"live": {TopicID: topicID}}
state.SessionStates["live"] = workspacestate.SessionState{Lifecycle: workspacestate.Archived}
if !topicHasSurvivingOwner(state, topicID) {
t.Fatal("archived session no longer owns its topic")
}
}
+50
View File
@@ -0,0 +1,50 @@
package main
import (
"strings"
"reasonix/internal/agent"
"reasonix/internal/control"
)
const sessionOperationHistoricalSourcePending = "historical_source_pending"
// nativeHistoricalSource names the transcript a native legacy runtime continues
// in place. That runtime has no durable submission store, so the source must
// be prepared into a canonical session before it can accept an identified send.
func nativeHistoricalSource(ctrl control.SessionAPI, path, headID string) *SessionSourceRef {
native, ok := ctrl.(*control.Controller)
path = strings.TrimSpace(path)
if !ok || path == "" || !native.NativeLegacySession() {
return nil
}
path = agent.CanonicalSessionPath(path)
return &SessionSourceRef{HostID: localDesktopHostID, Path: path, HeadID: headID, SourceKey: desktopSourceKey(path, headID)}
}
// tabHistoricalSourceLocked requires a.mu; it reports a source whether the tab
// was restored as a preparation shell or runs the source natively.
func tabHistoricalSourceLocked(tab *WorkspaceTab) *SessionSourceRef {
if tab == nil {
return nil
}
if tab.HistoricalSource != nil {
return tab.HistoricalSource
}
return nativeHistoricalSource(tab.Ctrl, tab.currentSessionPath(), tab.SessionHeadID)
}
// identifiedSubmissionCheck runs under the turn admission locks, so the
// refusal names the same runtime the submission would have reached.
func identifiedSubmissionCheck(submissionID []string) func(control.SessionAPI) error {
if firstSubmissionID(submissionID) == "" {
return nil
}
return func(ctrl control.SessionAPI) error {
if nativeHistoricalSource(ctrl, ctrl.SessionPath(), "") == nil {
return nil
}
return newSessionOperationError(sessionOperationHistoricalSourcePending,
"This conversation is still a historical source. Prepare it as a session before sending new messages.")
}
}
+115
View File
@@ -0,0 +1,115 @@
package main
import (
"errors"
"os"
"path/filepath"
"testing"
"time"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/control"
"reasonix/internal/provider"
)
func restoreHistoricalSourcePendingTab(t *testing.T) (*App, *WorkspaceTab, string) {
t.Helper()
app := newSavedTabReconcileTestApp(t)
t.Cleanup(app.closeSessionServices)
path := filepath.Join(config.SessionDir(), "pending.jsonl")
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
t.Fatal(err)
}
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "old", Role: provider.RoleUser, Content: "original work"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
entry := desktopTabEntry{ID: "pending", Scope: "global", TopicID: "topic", SessionPath: path}
file := desktopTabsFile{Tabs: []desktopTabEntry{entry}, ActiveTab: entry.ID, TabOrder: []string{entry.ID}}
if err := os.MkdirAll(desktopConfigDir(), 0o700); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(desktopConfigDir(), tabsFileName), mustMarshalJSON(t, file), 0o600); err != nil {
t.Fatal(err)
}
finishSavedTabMigration(app)
app.tabsRestored = make(chan struct{})
app.restoreOrBuildTabs()
tab := waitForTabReady(t, app, entry.ID)
native, ok := tab.Ctrl.(*control.Controller)
if !ok || !native.NativeLegacySession() || agent.CanonicalSessionPath(tab.SessionPath) != agent.CanonicalSessionPath(path) {
t.Fatalf("pending source did not restore as its native runtime: path=%q ctrl=%T", tab.SessionPath, tab.Ctrl)
}
return app, tab, path
}
func TestHistoricalSourcePendingTabRefusesSendWithTypedReason(t *testing.T) {
app, tab, path := restoreHistoricalSourcePendingTab(t)
err := app.SubmitToTabWithID(tab.ID, "continue", "submission-1")
var refusal *SessionOperationError
if !errors.As(err, &refusal) || refusal.Code != sessionOperationHistoricalSourcePending {
t.Fatalf("send was not refused with the historical-source reason: %v", err)
}
if !errors.Is(err, control.ErrSubmissionNotAccepted) {
t.Fatalf("refusal does not say the submission was not accepted: %v", err)
}
if errors.Is(err, control.ErrSubmissionIdentityUnavailable) {
t.Fatalf("send reached the controller before being refused: %v", err)
}
if tab.Ctrl.RuntimeStatus().Running {
t.Fatal("refused send started a turn")
}
meta := app.metaForTab(tab.ID)
if meta.HistoricalSource == nil || agent.CanonicalSessionPath(meta.HistoricalSource.Path) != agent.CanonicalSessionPath(path) {
t.Fatalf("meta does not report the pending source, so the composer stays enabled: %+v", meta.HistoricalSource)
}
for _, listed := range app.ListTabs() {
if listed.ID == tab.ID && listed.HistoricalSource == nil {
t.Fatal("tab list does not report the pending source")
}
}
}
func TestHistoricalSourcePendingTabPreparesIntoCanonicalSession(t *testing.T) {
app, tab, _ := restoreHistoricalSourcePendingTab(t)
source := app.metaForTab(tab.ID).HistoricalSource
if source == nil {
t.Fatal("meta does not report the pending source")
}
view, err := app.PrepareSession(SessionSelector{Source: source})
if err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(10 * time.Second)
for view.Status != "ready" && view.Status != "failed" && view.Status != "blocked" && time.Now().Before(deadline) {
time.Sleep(20 * time.Millisecond)
if view, err = app.GetSessionPreparation(view.OperationID); err != nil {
t.Fatal(err)
}
}
if view.Status != "ready" || view.Target == nil {
t.Fatalf("pending source could not be prepared while its tab is open: %+v", view)
}
}
func TestNativeLegacyRuntimeRefusesIdentifiedSendBeforeController(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "native.jsonl")
ctrl := control.New(control.Options{Sink: &tabEventSink{tabID: "tab"}, SessionDir: dir, SessionPath: path, NativeLegacySession: true})
defer ctrl.Close()
tab := &WorkspaceTab{ID: "tab", Scope: "global", Ready: true, Ctrl: ctrl, sink: &tabEventSink{tabID: "tab"}, SessionPath: path}
app := &App{tabs: map[string]*WorkspaceTab{tab.ID: tab}, activeTabID: tab.ID}
err := app.SubmitToTabWithID(tab.ID, "continue", "submission-1")
var refusal *SessionOperationError
if !errors.As(err, &refusal) || refusal.Code != sessionOperationHistoricalSourcePending {
t.Fatalf("send was not refused with the historical-source reason: %v", err)
}
if errors.Is(err, control.ErrSubmissionIdentityUnavailable) {
t.Fatalf("send reached the controller before being refused: %v", err)
}
}
+8 -1
View File
@@ -387,6 +387,7 @@ func commitOperation(state *State, id string, visiting map[string]bool) error {
}
entry.Status, entry.SessionID = "restored", op.SessionIDs[0]
state.RecoveryEntries[entry.ID] = entry
settleRecoveryVersion(state, entry)
}
op.Phase, op.ResultGeneration = "committed", state.Generation+1
state.PendingOperations[id] = op
@@ -639,6 +640,9 @@ func (s *Store) EnsureSessionTopic(ctx context.Context, id, topicID, title strin
func (s *Store) RecordRecovery(ctx context.Context, entry RecoveryEntry) error {
return s.mutate(ctx, func(state *State) error {
if _, restored := restoredRecoveryVersion(*state, entry.SourceKey, entry.Fingerprint); restored {
return nil
}
if old, ok := state.RecoveryEntries[entry.ID]; ok {
if old.Status == "restored" && old.Fingerprint == entry.Fingerprint {
return nil
@@ -703,7 +707,10 @@ func (s *Store) ReconcileDiscoveredSession(ctx context.Context, entry RecoveryEn
if !found {
workspaceID = workspace.ID
if existing, exists := state.Workspaces[workspaceID]; exists && existing.Root != workspace.Root {
return ErrMutationConflict
if workspaceID != GlobalWorkspaceID || strings.TrimSpace(workspace.Root) == "" {
return ErrMutationConflict
}
rebindGlobalRoot(state, workspace.Root, time.Now().UTC())
}
}
value, ok := state.Workspaces[workspaceID]
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"path/filepath"
"slices"
"strings"
"reasonix/internal/pathidentity"
@@ -129,6 +130,7 @@ func (r *ReadSnapshot) Generation() uint64 { return r.state.Generation }
func (r *ReadSnapshot) WorkspaceMetadata(id string) (Workspace, bool) {
w, ok := r.state.Workspaces[id]
w.SessionIDs, w.Organization, w.extra = nil, nil, nil
w.FormerRoots = slices.Clone(w.FormerRoots)
return w, ok
}
@@ -0,0 +1,55 @@
package workspacestate
import "context"
// restoredRecoveryVersion reports the session a source version was restored
// into. Entries are keyed by version, so the version, not whichever entry first
// reported it, decides whether recovery is still owed.
func restoredRecoveryVersion(state State, sourceKey, fingerprint string) (string, bool) {
if fingerprint == "" {
return "", false
}
found, sessionID := "", ""
for id, entry := range state.RecoveryEntries {
if entry.Status == "restored" && entry.SessionID != "" && entry.SourceKey == sourceKey &&
entry.Fingerprint == fingerprint && (found == "" || id < found) {
found, sessionID = id, entry.SessionID
}
}
return sessionID, found != ""
}
func settleRecoveryVersion(state *State, restored RecoveryEntry) {
if restored.Fingerprint == "" {
return
}
for id, entry := range state.RecoveryEntries {
if entry.Status != "restored" && entry.SourceKey == restored.SourceKey && entry.Fingerprint == restored.Fingerprint {
entry.Status, entry.SessionID = "restored", restored.SessionID
state.RecoveryEntries[id] = entry
}
}
}
// RebaseRecovery points a pending entry at the source version now on disk; the
// caller holds the source lock until its restore commits. A version already
// restored settles the entry to that session instead of importing it again.
func (s *Store) RebaseRecovery(ctx context.Context, id, fingerprint string) (RecoveryEntry, error) {
var rebased RecoveryEntry
err := s.mutate(ctx, func(state *State) error {
entry, ok := state.RecoveryEntries[id]
if !ok {
return ErrMutationConflict
}
if entry.Status != "restored" {
entry.Fingerprint = fingerprint
if sessionID, restored := restoredRecoveryVersion(*state, entry.SourceKey, fingerprint); restored {
entry.Status, entry.SessionID = "restored", sessionID
}
state.RecoveryEntries[id] = entry
}
rebased = entry
return nil
})
return rebased, err
}
@@ -22,6 +22,7 @@ func cloneSnapshot(s State) State {
s.Workspaces = maps.Clone(s.Workspaces)
for id, w := range s.Workspaces {
w.SessionIDs = slices.Clone(w.SessionIDs)
w.FormerRoots = slices.Clone(w.FormerRoots)
w.extra = cloneUnknownFields(w.extra)
if w.Organization != nil {
o := w.Organization.Clone()
+2 -1
View File
@@ -34,6 +34,7 @@ type Workspace struct {
Organization *Organization `json:"organization,omitempty"`
ID string `json:"id"`
Root string `json:"root"`
FormerRoots []string `json:"formerRoots,omitempty"` // global only: roots it was rebound away from
Title string `json:"title"`
SessionIDs []string `json:"sessionIds"`
Visible bool `json:"visible"`
@@ -663,7 +664,7 @@ func (w *Workspace) UnmarshalJSON(body []byte) error {
if err := json.Unmarshal(body, &fields); err != nil {
return err
}
for _, key := range []string{"id", "root", "title", "sessionIds", "visible", "createdAt", "updatedAt", "organization"} {
for _, key := range []string{"id", "root", "title", "sessionIds", "visible", "createdAt", "updatedAt", "organization", "formerRoots"} {
delete(fields, key)
}
*w = Workspace(decoded)
@@ -235,6 +235,13 @@ func (s *Store) EnsureWorkspaceResolved(ctx context.Context, workspace Workspace
if current.Root == workspace.Root || (current.Root == "" && workspace.Root == "") {
return revalidateCandidate()
}
if workspace.ID == GlobalWorkspaceID && candidate.Key != "" {
if err := revalidateCandidate(); err != nil {
return err
}
rebindGlobalRoot(state, workspace.Root, now)
return nil
}
if workspace.ID == GlobalWorkspaceID || candidate.Key == "" {
return ErrMutationConflict
}
@@ -266,6 +273,19 @@ func (s *Store) EnsureWorkspaceResolved(ctx context.Context, workspace Workspace
return resolvedID, err
}
// rebindGlobalRoot moves global to root. Global is a singleton derived from the
// state root, so a different stored root is where the data directory used to
// be, not a second workspace; it is kept so sessions recorded there stay global.
func rebindGlobalRoot(state *State, root string, now time.Time) {
w := state.Workspaces[GlobalWorkspaceID]
former := slices.DeleteFunc(slices.Clone(w.FormerRoots), func(r string) bool { return r == root || r == w.Root })
if w.Root != "" {
former = append(former, w.Root)
}
w.Root, w.FormerRoots, w.UpdatedAt = root, former, now
state.Workspaces[GlobalWorkspaceID] = w
}
func versionedWorkspaceID(identityKey string) string {
sum := sha256.Sum256([]byte("reasonix-workspace-pathidentity-v2\x00" + identityKey))
return "project-v2-" + hex.EncodeToString(sum[:12])
@@ -247,17 +247,71 @@ func TestEnsureWorkspaceRejectsDifferentRootWithoutRewritingState(t *testing.T)
if err != nil {
t.Fatal(err)
}
roots := []string{"", filepath.Join(root, "different")}
if runtime.GOOS != "windows" && runtime.GOOS != "darwin" {
roots = append(roots, strings.ToUpper(root))
}
for _, other := range roots {
if err := store.EnsureWorkspace(t.Context(), Workspace{ID: GlobalWorkspaceID, Root: other}); !errors.Is(err, ErrMutationConflict) {
t.Fatalf("different root %q: %v", other, err)
}
if err := store.EnsureWorkspace(t.Context(), Workspace{ID: GlobalWorkspaceID, Root: ""}); !errors.Is(err, ErrMutationConflict) {
t.Fatalf("empty root: %v", err)
}
after, err := os.ReadFile(store.Path())
if err != nil || string(after) != string(before) {
t.Fatalf("conflicting root rewrote persisted state: %v", err)
}
}
// Global is derived from the data directory, so a stored root elsewhere is
// where that directory used to be. Refusing it left Global unloadable after a
// portable build or REASONIX_HOME moved (#10638, #10659).
func TestEnsureWorkspaceResolvedRebindsGlobalAfterDataDirectoryMoves(t *testing.T) {
oldRoot := filepath.Join(t.TempDir(), "old-home", "global-workspace")
newRoot := filepath.Join(t.TempDir(), "new-home", "global-workspace")
path := filepath.Join(t.TempDir(), "workspace-state-v1.json")
state := newState()
state.WorkspaceIDs = []string{GlobalWorkspaceID}
state.Workspaces[GlobalWorkspaceID] = Workspace{ID: GlobalWorkspaceID, Root: oldRoot, Title: "Global", Visible: true, SessionIDs: []string{"kept"}}
state.SessionStates = map[string]SessionState{"kept": {Lifecycle: Active}}
writeRegistryState(t, path, state)
id, err := NewStore(path).EnsureWorkspaceResolved(t.Context(), Workspace{ID: GlobalWorkspaceID, Root: newRoot, Title: "Global", Visible: true})
if err != nil || id != GlobalWorkspaceID {
t.Fatalf("resolved id = %q, err = %v", id, err)
}
loaded, err := NewStore(path).Load(t.Context())
if err != nil {
t.Fatal(err)
}
global := loaded.Workspaces[GlobalWorkspaceID]
if global.Root != newRoot || !slices.Equal(global.FormerRoots, []string{oldRoot}) || !slices.Equal(global.SessionIDs, []string{"kept"}) {
t.Fatalf("global after rebind = %+v", global)
}
if len(loaded.WorkspaceIDs) != 1 {
t.Fatalf("rebind registered another workspace: %v", loaded.WorkspaceIDs)
}
if _, err := NewStore(path).EnsureWorkspaceResolved(t.Context(), Workspace{ID: GlobalWorkspaceID, Root: oldRoot}); err != nil {
t.Fatalf("moving back: %v", err)
}
if back, _ := NewStore(path).Load(t.Context()); back.Workspaces[GlobalWorkspaceID].Root != oldRoot || !slices.Equal(back.Workspaces[GlobalWorkspaceID].FormerRoots, []string{newRoot}) {
t.Fatalf("global after moving back = %+v", back.Workspaces[GlobalWorkspaceID])
}
}
func TestReconcileDiscoveredSessionRebindsGlobalAfterDataDirectoryMoves(t *testing.T) {
oldRoot := filepath.Join(t.TempDir(), "old-home", "global-workspace")
newRoot := filepath.Join(t.TempDir(), "new-home", "global-workspace")
path := filepath.Join(t.TempDir(), "workspace-state-v1.json")
state := newState()
state.WorkspaceIDs = []string{GlobalWorkspaceID}
state.Workspaces[GlobalWorkspaceID] = Workspace{ID: GlobalWorkspaceID, Root: oldRoot, Visible: true}
writeRegistryState(t, path, state)
err := NewStore(path).ReconcileDiscoveredSession(t.Context(),
RecoveryEntry{ID: "canonical-s", SourceKey: "canonical:s", SessionID: "s", Format: "canonical", Status: "pending"},
&Workspace{ID: GlobalWorkspaceID, Root: newRoot, Visible: true})
if err != nil {
t.Fatal(err)
}
loaded, err := NewStore(path).Load(t.Context())
if err != nil {
t.Fatal(err)
}
if global := loaded.Workspaces[GlobalWorkspaceID]; global.Root != newRoot || !slices.Equal(global.SessionIDs, []string{"s"}) {
t.Fatalf("global after discovery = %+v", global)
}
}
+15 -1
View File
@@ -6,6 +6,7 @@ import (
"reasonix/internal/agent"
"reasonix/internal/boot"
"reasonix/internal/control"
"reasonix/internal/event"
"reasonix/internal/evidence"
goaldomain "reasonix/internal/goal"
@@ -18,6 +19,11 @@ func (a *App) metaForTab(tabID string) Meta {
tab := a.tabByIDLocked(tabID)
snap := snapshotTabRuntimeLocked(tab)
runtimeView := a.sessionRuntimeViewLocked(tab)
var historical *SessionSourceRef
headID := ""
if tab != nil {
historical, headID = tab.HistoricalSource, tab.SessionHeadID
}
a.mu.RUnlock()
if tab == nil {
meta := Meta{EventChannel: eventChannel}
@@ -56,6 +62,14 @@ func (a *App) metaForTab(tabID string) Meta {
}
}
sessionPath := strings.TrimSpace(snap.sessionPath)
// Read off the lock, like every other controller read in this function.
if native, ok := snap.ctrl.(*control.Controller); ok && historical == nil {
path := sessionPath
if path == "" {
path = native.SessionPath()
}
historical = nativeHistoricalSource(native, path, headID)
}
sessionID := strings.TrimSpace(snap.sessionID)
var sessionRef *session.SessionRef
if sessionID != "" {
@@ -71,7 +85,7 @@ func (a *App) metaForTab(tabID string) Meta {
meta := Meta{
Label: snap.label, Ready: runtimeView.Phase == sessionRuntimeReady && snap.ctrl != nil,
Runtime: runtimeView, StartupErr: snap.startupErr, EventChannel: eventChannel,
HistoricalSource: snap.historicalSource,
HistoricalSource: historical,
SessionPath: sessionPath, SessionID: sessionID, Session: sessionRef,
SessionRevision: sessionRevision, SessionDigest: sessionDigest,
SessionGeneration: snap.sessionGeneration, RuntimeStateSnapshot: runtimeStateSnapshot,
+1 -1
View File
@@ -21,7 +21,7 @@ func (a *App) startModelApplicationTurn(tabID, submissionID string, req control.
if receipt, found, err := a.knownSubmissionReceipt(tabID, req); found || err != nil {
return TurnStartView{TurnID: receipt.TurnID, SubmissionID: submissionID, Status: event.TurnQueued, Disposition: control.SubmitTurnStarted}, err
}
admission, ctrl, err := a.beginRuntimeTurnWithModelChoice(tabID, true, false, nil, choice, submissionID)
admission, ctrl, err := a.beginRuntimeTurnWithModelChoice(tabID, true, false, identifiedSubmissionCheck([]string{submissionID}), choice, submissionID)
if err != nil {
return TurnStartView{}, a.submissionAdmissionError(tabID, req, err)
}
+2 -2
View File
@@ -341,7 +341,7 @@ func (a *App) stageArchiveSource(ctx context.Context, path string) (session.Sess
return session.SessionRef{}, "", err
}
scope, root := "global", ""
if meta.WorkspaceRoot != "" && !sameDesktopPath(meta.WorkspaceRoot, globalWorkspaceRoot()) {
if meta.WorkspaceRoot != "" && !a.isGlobalWorkspacePath(ctx, meta.WorkspaceRoot) {
scope, root = "project", meta.WorkspaceRoot
}
workspaceID, err := a.ensureDesktopWorkspace(ctx, scope, root)
@@ -381,7 +381,7 @@ func (a *App) restoreCanonicalSession(ctx context.Context, ref session.SessionRe
return SessionRestoreResult{}, errSessionWorkspaceConflict
}
scope, root := "project", info.CWD
if sameDesktopPath(root, globalWorkspaceRoot()) {
if a.isGlobalWorkspacePath(ctx, root) {
scope, root = "global", ""
}
workspaceID, ensureErr := a.ensureDesktopWorkspace(ctx, scope, root)
+16
View File
@@ -0,0 +1,16 @@
package main
// selectLocalSurfaceAfterOpen makes a bound local session the visible surface
// even when a remote tab was selected before the request. It serializes with
// direct tab clicks, so a newer selection is never cleared by this completion.
func (a *App) selectLocalSurfaceAfterOpen(navigationSequence uint64) {
a.tabSelectionMu.Lock()
defer a.tabSelectionMu.Unlock()
if a.desktopSessions.navigationSeq.Load() != navigationSequence {
return
}
a.remoteTabMu.Lock()
a.remoteTabLayout.activeID = ""
a.remoteTabMu.Unlock()
a.queueCurrentTabLayout()
}
+13 -6
View File
@@ -38,7 +38,7 @@ func (a *App) recoveryWorkspaceChoices(ctx context.Context, state workspacestate
if entry.SessionID != "" {
if info, err := a.desktopSessionService("").Query().Stat(ctx, session.SessionRef{HostID: localDesktopHostID, SessionID: entry.SessionID}); err == nil && info.CWD != "" {
for id, w := range state.Workspaces {
if sameDesktopPath(w.Root, info.CWD) {
if sameDesktopPath(w.Root, info.CWD) || (id == workspacestate.GlobalWorkspaceID && isGlobalWorkspacePath(w, info.CWD)) {
allowed[id] = true
}
}
@@ -275,8 +275,15 @@ func (a *App) restoreRecoveryEntryInWorkspace(id, operationID, workspaceID strin
if err != nil {
return SessionRestoreResult{}, err
}
if entry.Fingerprint != "" && entry.Fingerprint != fingerprint {
return SessionRestoreResult{}, errors.New("historical source changed; rescan before restoring")
if entry.Fingerprint != fingerprint {
rebased, err := a.workspaceRegistry().RebaseRecovery(ctx, id, fingerprint)
if err != nil {
return SessionRestoreResult{}, err
}
if rebased.SessionID != "" {
return a.restoreCanonicalSession(ctx, session.SessionRef{HostID: localDesktopHostID, SessionID: rebased.SessionID}, operationID, id)
}
entry.Fingerprint = rebased.Fingerprint
}
workspaceID, err = a.ensureDesktopWorkspace(ctx, entry.Scope, entry.WorkspaceRoot)
if err != nil {
@@ -329,7 +336,7 @@ func (a *App) discoverHistoricalTrash(ctx context.Context) error {
}
for _, path := range paths {
scope, root := "global", ""
if meta, ok, err := agent.LoadBranchMeta(path); err == nil && ok && meta.WorkspaceRoot != "" && !sameDesktopPath(meta.WorkspaceRoot, globalWorkspaceRoot()) {
if meta, ok, err := agent.LoadBranchMeta(path); err == nil && ok && meta.WorkspaceRoot != "" && !a.isGlobalWorkspacePath(ctx, meta.WorkspaceRoot) {
scope, root = "project", meta.WorkspaceRoot
}
// Old "deleted" entries were recoverable trash, not permanent
@@ -445,7 +452,7 @@ func (a *App) reconcileUnregisteredSessions(ctx context.Context) error {
continue
}
scope, root := "project", info.CWD
if sameDesktopPath(root, globalWorkspaceRoot()) {
if isGlobalWorkspacePath(state.Workspaces[workspacestate.GlobalWorkspaceID], root) {
scope, root = "global", ""
}
title := workspaceName(root)
@@ -610,7 +617,7 @@ func (a *App) restoreLegacyRecoveryPath(path string) error {
return fmt.Errorf("session is open: %s", key)
}
scope, root := "global", ""
if meta, ok, err := agent.LoadBranchMeta(path); err == nil && ok && meta.WorkspaceRoot != "" && !sameDesktopPath(meta.WorkspaceRoot, globalWorkspaceRoot()) {
if meta, ok, err := agent.LoadBranchMeta(path); err == nil && ok && meta.WorkspaceRoot != "" && !a.isGlobalWorkspacePath(a.bootContext(), meta.WorkspaceRoot) {
scope, root = "project", meta.WorkspaceRoot
}
if err := a.sourceRecovery(a.bootContext(), path, "legacy-trash", "historical_state_unknown", scope, root); err != nil {
+92
View File
@@ -0,0 +1,92 @@
package main
import (
"path/filepath"
"testing"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
)
func TestRestoreRecoveryUsesSourceVersionOnDiskAfterFurtherChange(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionDir()
sourceFile := filepath.Join(root, "old.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "old", Role: provider.RoleUser, Content: "original work"})
if err := legacy.Save(sourceFile); err != nil {
t.Fatal(err)
}
appendLine := func(content string) {
legacy.Add(provider.Message{ID: content, Role: provider.RoleUser, Content: content})
if err := legacy.Save(sourceFile); err != nil {
t.Fatal(err)
}
}
source := desktopMigrationSource{root: root, scope: "global", exact: map[string]bool{sourceFile: true}}
migrate := func(app *App) {
t.Helper()
if err := app.migrateLegacyDirectory(t.Context(), source); err != nil {
t.Fatal(err)
}
source.exact = nil
}
restart := func(app *App) *App {
app.closeSessionServices()
next := NewApp()
t.Cleanup(next.closeSessionServices)
return next
}
pending := func(app *App) []RecoveryEntryView {
t.Helper()
page, err := app.ListRecoveryEntries("", "", 50)
if err != nil {
t.Fatal(err)
}
return page.Items
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
migrate(app)
appendLine("first-change")
app = restart(app)
migrate(app)
entries := pending(app)
if len(entries) != 1 || entries[0].Reason != "source_changed_after_adoption" {
t.Fatalf("changed source was not quarantined: %+v", entries)
}
quarantined := entries[0].ID
appendLine("second-change")
app = restart(app)
migrate(app)
release, err := acquireHistoricalSource(t.Context(), desktopSourceKey(sourceFile, ""), historicalSource{path: sourceFile, format: "legacy"})
if err != nil {
t.Fatal(err)
}
if _, err := app.RestoreRecoveryEntry(quarantined, ""); !historicalSourceBusyError(err) {
t.Fatalf("restore did not refuse a source another writer holds: %v", err)
}
release()
result, err := app.RestoreRecoveryEntry(quarantined, "")
if err != nil {
t.Fatalf("restore refused a source that changed again: %v", err)
}
history, err := app.desktopSessionService("").Query().History(t.Context(), session.SessionRef{HostID: localDesktopHostID, SessionID: result.Session.SessionID})
if err != nil || len(history) == 0 || history[len(history)-1].Content != "second-change" {
t.Fatalf("restore did not read the version on disk: %+v %v", history, err)
}
if left := pending(app); len(left) != 0 {
t.Fatalf("restored version is still offered for recovery: %+v", left)
}
app = restart(app)
for range 2 {
migrate(app)
}
if left := pending(app); len(left) != 0 {
t.Fatalf("restart re-quarantined the restored version: %+v", left)
}
}
+1
View File
@@ -723,6 +723,7 @@ func (a *App) openSessionWithNavigation(ref session.SessionRef, navigationSequen
}
return HistoryPage{}, err
}
a.selectLocalSurfaceAfterOpen(navigationSequence)
// runtime:rebuilt intentionally has no reload semantics. SessionRef opening
// is navigation, so publish ready only after the exact target commits and
// let every frontend owner re-read its metadata and history.
+1
View File
@@ -103,6 +103,7 @@ func (a *App) canonicalSessionWorkspace(ctx context.Context, ref session.Session
if identityErr != nil {
return owner, identityErr
}
same = same || (owner.ID == workspacestate.GlobalWorkspaceID && isGlobalWorkspacePath(owner, info.CWD))
if !same {
return owner, errSessionWorkspaceConflict
}
@@ -11,6 +11,32 @@ import (
"reasonix/internal/session"
)
func TestOpenSessionSelectsLocalTabAfterRemote(t *testing.T) {
app, _, target, _, _ := canonicalWorkspaceOpenFixture(t)
app.remoteTabMu.Lock()
app.remoteTabs = map[string]*remoteTab{
"remote": {id: "remote", ref: RemoteTabRef{HostID: "box", Workspace: "/work"}, state: "disconnected"},
}
app.remoteTabLayout = remoteTabLayoutState{activeID: "remote", order: []string{"remote"}, stripOrder: []string{"remote"}}
app.remoteTabMu.Unlock()
if _, err := app.OpenSession(target.Ref()); err != nil {
t.Fatal(err)
}
localActive, remoteActive := false, false
for _, tab := range app.ListTabs() {
if tab.Remote == nil && tab.SessionID == target.Ref().SessionID {
localActive = tab.Active
}
if tab.Remote != nil {
remoteActive = tab.Active
}
}
if !localActive || remoteActive {
t.Fatalf("OpenSession left localActive=%v remoteActive=%v", localActive, remoteActive)
}
}
func TestCanonicalNavigationLastRequestWins(t *testing.T) {
app, tab, target, _, _ := canonicalWorkspaceOpenFixture(t)
ctrl := tab.Ctrl
+18
View File
@@ -8,6 +8,7 @@ import (
"fmt"
"log/slog"
"path/filepath"
"slices"
"strings"
"sync"
"sync/atomic"
@@ -155,6 +156,23 @@ func desktopWorkspaceRoot(scope, workspaceRoot string) string {
return filepath.Clean(strings.TrimSpace(workspaceRoot))
}
// isGlobalWorkspacePath reports whether root is Global's directory: the one
// derived from the data directory now, or one Global was rebound away from.
func isGlobalWorkspacePath(global workspacestate.Workspace, root string) bool {
if sameDesktopPath(root, globalWorkspaceRoot()) || (global.Root != "" && sameDesktopPath(root, global.Root)) {
return true
}
return slices.ContainsFunc(global.FormerRoots, func(former string) bool { return sameDesktopPath(root, former) })
}
func (a *App) isGlobalWorkspacePath(ctx context.Context, root string) bool {
if sameDesktopPath(root, globalWorkspaceRoot()) {
return true
}
state, err := a.workspaceRegistry().Load(ctx)
return err == nil && isGlobalWorkspacePath(state.Workspaces[workspacestate.GlobalWorkspaceID], root)
}
func (a *App) workspaceRegistry() *workspacestate.Store {
if a == nil {
return nil
@@ -167,3 +167,44 @@ func TestWorkspaceRegistryRejectsSessionHeaderFromDifferentWorkspace(t *testing.
t.Fatalf("mismatched workspace members = %#v", workspace.SessionIDs)
}
}
// A portable build or REASONIX_HOME move leaves Global's stored root and every
// global session's recorded CWD at the old data directory (#10638, #10659).
func TestGlobalWorkspaceSurvivesMovedDataDirectory(t *testing.T) {
isolateDesktopUserDirs(t)
root := t.TempDir()
oldGlobal := filepath.Join(t.TempDir(), "old-home", "global-workspace")
app := NewApp()
t.Cleanup(app.closeSessionServices)
app.ctx = t.Context()
app.desktopSessions.root = filepath.Join(root, "desktop-sessions-v5", "by-id")
app.desktopSessions.workspaceState = workspacestate.NewStore(filepath.Join(root, "desktop", "workspace-state-v1.json"))
store := app.desktopSessions.workspaceState
if err := store.EnsureWorkspace(t.Context(), workspacestate.Workspace{ID: workspacestate.GlobalWorkspaceID, Root: oldGlobal, Visible: true}); err != nil {
t.Fatal(err)
}
created, err := app.desktopSessionService("").Create(t.Context(), session.CreateOptions{
SessionID: "moved-global", CWD: oldGlobal, Origin: session.SessionOriginNew,
})
if err != nil {
t.Fatal(err)
}
if err := store.AttachSession(t.Context(), "", workspacestate.GlobalWorkspaceID, created.Ref().SessionID, ""); err != nil {
t.Fatal(err)
}
if id, err := app.ensureDesktopWorkspace(t.Context(), "global", ""); err != nil || id != workspacestate.GlobalWorkspaceID {
t.Fatalf("ensure global = %q, %v", id, err)
}
state, err := store.Load(t.Context())
if err != nil {
t.Fatal(err)
}
if got := state.Workspaces[workspacestate.GlobalWorkspaceID].Root; got != globalWorkspaceRoot() {
t.Fatalf("global root = %q, want %q", got, globalWorkspaceRoot())
}
owner, err := app.canonicalSessionWorkspace(t.Context(), created.Ref())
if err != nil || owner.ID != workspacestate.GlobalWorkspaceID {
t.Fatalf("session recorded under the old global root = %q, %v", owner.ID, err)
}
}
+17
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"sync"
)
@@ -83,10 +84,26 @@ func (a *App) pruneVisibleTabsRuntimeAdmissionHeld(tabID string) (TabMeta, error
a.removeTabOrderLocked(id)
}
a.tabOrder = []string{tabID}
// A local topic replaces the remote visible surface too. Leaving its
// registry entry active makes ListTabs hide the newly selected local tab.
a.remoteTabMu.Lock()
remoteCancels := make([]context.CancelFunc, 0, len(a.remoteTabs))
for id, remote := range a.remoteTabs {
if remote.cancel != nil {
remoteCancels = append(remoteCancels, remote.cancel)
}
delete(a.remoteTabs, id)
a.forgetRemoteBrowserExecutor(id)
}
a.remoteTabLayout = remoteTabLayoutState{}
a.remoteTabMu.Unlock()
a.saveTabsLocked()
meta := a.tabMeta(active, true)
a.mu.Unlock()
for _, cancel := range remoteCancels {
cancel()
}
for _, tab := range removed {
a.removeVisibleTabRuntimeAdmissionHeld(tab)
}
+1 -3
View File
@@ -11,7 +11,6 @@ type tabRuntimeSnapshot struct {
ready bool
readOnly bool
startupErr string
historicalSource *SessionSourceRef
scope string
workspaceRoot string
sessionPath string
@@ -43,8 +42,7 @@ func snapshotTabRuntimeLocked(tab *WorkspaceTab) tabRuntimeSnapshot {
return tabRuntimeSnapshot{
ctrl: tab.Ctrl, sink: tab.sink, label: tab.Label, ready: tab.Ready,
readOnly: tab.ReadOnly, startupErr: tab.StartupErr, scope: tab.Scope,
historicalSource: tab.HistoricalSource,
workspaceRoot: tab.WorkspaceRoot, sessionPath: tab.SessionPath, sessionID: tab.SessionID,
workspaceRoot: tab.WorkspaceRoot, sessionPath: tab.SessionPath, sessionID: tab.SessionID,
sessionGeneration: tab.SessionGeneration, topicID: tab.TopicID, topicTitle: tab.TopicTitle,
sharedHostKey: tab.SharedHostKey, model: tab.model, effort: cloneStringPtr(tab.effort),
tokenMode: currentTabTokenMode(tab), qualityFloor: tab.qualityFloor, mode: tab.mode,
+1 -1
View File
@@ -2187,7 +2187,7 @@ func (a *App) tabMeta(tab *WorkspaceTab, active bool) TabMeta {
Goal: currentTabGoal(tab),
GoalStatus: currentTabGoalStatus(tab),
StartupErr: tab.StartupErr,
HistoricalSource: tab.HistoricalSource,
HistoricalSource: tabHistoricalSourceLocked(tab),
Active: active,
Cwd: tab.WorkspaceRoot,
IsolatedWorktree: floor.isolated,
+6
View File
@@ -213,6 +213,12 @@ func (a *App) StartTopicActivation(req TopicActivationRequest) (TopicActivationT
// untouched on error.
return TopicActivationTicket{}, err
}
// The ticket publishes the local surface before the asynchronous prune.
// Clear the remote selection now so ListTabs cannot reselect the old remote
// tab while the local controller is still becoming ready.
a.remoteTabMu.Lock()
a.remoteTabLayout.activeID = ""
a.remoteTabMu.Unlock()
requestID := strings.TrimSpace(req.RequestID)
if requestID == "" {
@@ -34,6 +34,53 @@ func canonicalTopicHistoryReady(t *testing.T, app *App, tabID string) session.Hi
}
}
func TestCanonicalSidebarActivationReplacesRemoteSurface(t *testing.T) {
app, source, target, root, _ := canonicalWorkspaceOpenFixture(t)
app.readyHook = func() {}
installNoopRuntimeEvents(app, source.sink)
events := newActivationEventRecorder(app)
t.Cleanup(func() { app.shutdown(context.Background()) })
remoteCtx, cancelRemote := context.WithCancel(context.Background())
app.remoteTabMu.Lock()
app.remoteTabs = map[string]*remoteTab{
"remote": {id: "remote", ref: RemoteTabRef{HostID: "box", Workspace: "/work"}, state: "ready", cancel: cancelRemote},
}
app.remoteTabLayout = remoteTabLayoutState{activeID: "remote", order: []string{"remote"}, stripOrder: []string{"remote"}}
app.remoteTabMu.Unlock()
ref := target.Ref()
ticket, err := app.StartTopicActivation(TopicActivationRequest{
Selector: &SessionSelector{Ref: &ref}, Scope: "project", WorkspaceRoot: root,
TopicID: "existing-topic", SessionPath: sessionRoute(ref.SessionID), RequestID: "sidebar-local",
})
if err != nil {
t.Fatal(err)
}
if ticket.Meta.SessionID != ref.SessionID || ticket.Meta.WorkspaceRoot != root {
t.Fatalf("selected local session = %+v, want %s in %s", ticket.Meta, ref.SessionID, root)
}
if tabs := app.ListTabs(); len(tabs) == 0 {
t.Fatal("local tab disappeared after canonical sidebar activation")
} else {
for _, tab := range tabs {
if tab.Active != (tab.ID == ticket.TabID) {
t.Fatalf("tab activity after canonical sidebar activation = %+v", tabs)
}
}
}
events.waitFor(t, activationEventFor(ticket.RequestID, "ready"))
flushActivationCompletions(app)
if tabs := app.ListTabs(); len(tabs) != 1 || tabs[0].ID != ticket.TabID || !tabs[0].Active {
t.Fatalf("tabs after local activation = %+v", tabs)
}
select {
case <-remoteCtx.Done():
default:
t.Fatal("remote tab remained attached after selecting a local session")
}
}
func TestCanonicalTopicActivationPublishesReadableHistoryBeforeRuntime(t *testing.T) {
app, source, target, root, workspace := canonicalWorkspaceOpenFixture(t)
app.readyHook = func() {}
+50
View File
@@ -209,6 +209,56 @@ func TestStartTopicActivationSyncBuildAndReuseFastPath(t *testing.T) {
}
}
func TestStartTopicActivationReplacesRemoteSurface(t *testing.T) {
isolateDesktopUserDirs(t)
app := NewApp()
app.ctx = context.Background()
app.readyHook = func() {}
installNoopRuntimeEvents(app)
events := newActivationEventRecorder(app)
gate := newTabBuildGate(app)
t.Cleanup(func() { gate.releaseAll(); app.shutdown(context.Background()) })
remoteCtx, cancelRemote := context.WithCancel(context.Background())
app.remoteTabMu.Lock()
app.remoteTabs = map[string]*remoteTab{
"remote": {id: "remote", ref: RemoteTabRef{HostID: "box", Workspace: "/work"}, state: "disconnected", cancel: cancelRemote},
}
app.remoteTabLayout = remoteTabLayoutState{activeID: "remote", order: []string{"remote"}, stripOrder: []string{"remote"}}
app.remoteTabMu.Unlock()
ticket, err := app.StartTopicActivation(TopicActivationRequest{Scope: "global", TopicID: "local-after-remote", RequestID: "local-after-remote"})
if err != nil {
t.Fatal(err)
}
// The tab build stays held: the local tab must be selected without it.
gate.waitEntered(t, ticket.TabID)
if tabs := app.ListTabs(); len(tabs) == 0 {
t.Fatalf("tabs immediately after local ticket = %+v, want active %q", tabs, ticket.TabID)
} else {
active := ""
for _, tab := range tabs {
if tab.Active {
active = tab.ID
}
}
if active != ticket.TabID {
t.Fatalf("tabs immediately after local ticket = %+v, want active %q", tabs, ticket.TabID)
}
}
gate.release(ticket.TabID)
events.waitFor(t, activationEventFor(ticket.RequestID, "ready"))
flushActivationCompletions(app)
if tabs := app.ListTabs(); len(tabs) != 1 || tabs[0].ID != ticket.TabID || !tabs[0].Active {
t.Fatalf("tabs after local activation = %+v, want only active local %q", tabs, ticket.TabID)
}
select {
case <-remoteCtx.Done():
default:
t.Fatal("remote tab pump was not cancelled after switching to local")
}
}
// activationStubController is the minimal SessionAPI surface exercised by
// prune/detach/attach flows around a tab with active runtime work.
type activationStubController struct {
+1 -1
View File
@@ -60,7 +60,7 @@ func (admission *tabTurnAdmission) abort() {
// beginTabTurn reserves one tab until its TurnDone fan-out completes.
func (a *App) beginTabTurn(tabID string, reclaim bool, submissionID ...string) (*tabTurnAdmission, control.SessionAPI, error) {
return a.beginRuntimeTurn(tabID, reclaim, false, submissionID...)
return a.beginRuntimeTurnChecked(tabID, reclaim, false, identifiedSubmissionCheck(submissionID), submissionID...)
}
func (a *App) beginRuntimeTurn(tabID string, reclaim, detached bool, submissionID ...string) (*tabTurnAdmission, control.SessionAPI, error) {
+12
View File
@@ -68,6 +68,18 @@ func ReadSessionHeadIndex(sessionPath string) (*SessionHeadIndex, error) {
return &idx, nil
}
// sessionHeadIndexStale reports a schema-2 log whose head index no longer
// describes it. The listing index cannot answer for heads, so only a replay
// that rewrites the head index repairs such a session.
func sessionHeadIndexStale(sessionPath string) bool {
probe, err := probeSessionEventLog(sessionPath)
if err != nil || !probe.dag {
return false
}
idx, err := ReadSessionHeadIndex(sessionPath)
return err != nil || !idx.Current(sessionPath)
}
func writeSessionDAGIndex(ctx context.Context, sessionPath string, st *sessionDAGState) error {
indexPath := store.SessionEventIndex(sessionPath)
if indexPath == "" || st == nil {
+4 -2
View File
@@ -88,8 +88,10 @@ func RepairSessionListingProjection(ctx context.Context, path string) (result Se
if !metaOK {
meta = BranchMeta{ID: BranchID(path)}
}
if result, handled, err := repairSessionListingFromIndex(path, meta); handled || err != nil {
return result, err
if !sessionHeadIndexStale(path) {
if result, handled, err := repairSessionListingFromIndex(path, meta); handled || err != nil {
return result, err
}
}
return repairSessionListingFromReplay(ctx, path, meta)
}
+20
View File
@@ -6,6 +6,7 @@ package boot
import (
"context"
"encoding/json"
"fmt"
"reflect"
"sync"
@@ -233,3 +234,22 @@ model = "x"
t.Fatal("no round reached the provider; the run never started")
}
}
// TestEffectEveryProviderToolDeclaresRequiredArray holds the root schema shape
// strict upstreams validate: relays re-serialize an omitted required as null.
func TestEffectEveryProviderToolDeclaresRequiredArray(t *testing.T) {
reqs := effectRun(t, "boot-effect-required", "", ablation.Set{})
if len(reqs[0].Tools) == 0 {
t.Fatal("no tools reached the provider request")
}
for _, schema := range reqs[0].Tools {
var root map[string]json.RawMessage
if err := json.Unmarshal(schema.Parameters, &root); err != nil {
t.Fatalf("%s: parameters are not an object: %v", schema.Name, err)
}
var required []string
if raw, ok := root["required"]; !ok || string(raw) == "null" || json.Unmarshal(raw, &required) != nil {
t.Errorf("%s: parameters lack a required array: %s", schema.Name, schema.Parameters)
}
}
}
+3 -3
View File
@@ -1,8 +1,8 @@
{
"SystemHash": "2c95f2f47dec3aed",
"ToolsHash": "ca8675322833ed59",
"PrefixHash": "5b156eb122da7ac2",
"ToolsHash": "69878e44e84d9a93",
"PrefixHash": "29044abb555d147e",
"LogRewriteVersion": 0,
"ToolSchemaTokens": 3406,
"ToolSchemaTokens": 3409,
"SessionContextDigest": ""
}
+1
View File
@@ -245,6 +245,7 @@
"parameters": {
"additionalProperties": false,
"properties": {},
"required": [],
"type": "object"
}
},
+1
View File
@@ -221,6 +221,7 @@
"Schema": {
"additionalProperties": false,
"properties": {},
"required": [],
"type": "object"
}
},
@@ -1,8 +1,8 @@
{
"SystemHash": "2c95f2f47dec3aed",
"ToolsHash": "25329d2755c20503",
"PrefixHash": "b9085bb6a2b359e8",
"ToolsHash": "0536a510c9605724",
"PrefixHash": "c2094426e8aacc2d",
"LogRewriteVersion": 0,
"ToolSchemaTokens": 3477,
"ToolSchemaTokens": 3480,
"SessionContextDigest": ""
}
@@ -193,6 +193,7 @@
"parameters": {
"additionalProperties": false,
"properties": {},
"required": [],
"type": "object"
}
},
@@ -168,6 +168,7 @@
"Schema": {
"additionalProperties": false,
"properties": {},
"required": [],
"type": "object"
}
},
+3 -1
View File
@@ -104,8 +104,10 @@ func HasSavedAccount(accountID string) bool {
}
// validAccountID keeps an id from the upstream login response to one file name.
// validAccountID accepts one file name. Either separator is refused on every
// OS: the config directory can be read from a system where the other one is.
func validAccountID(accountID string) bool {
return filepath.IsLocal(accountID) && filepath.Base(accountID) == accountID
return !strings.ContainsAny(accountID, `/\`) && filepath.IsLocal(accountID) && filepath.Base(accountID) == accountID
}
func saveAccount(accountID string, account savedAccount) error {
+6 -5
View File
@@ -5217,12 +5217,13 @@ func (c *Controller) applyToolApprovalModeLocked(mode string) []string {
p.reply <- approvalReply{allow: true}
drained = append(drained, p.id)
}
// A permission revision change invalidates the active turn so an older
// approval can never authorize work under the new snapshot. Avoid the idle
// Cancel path: it intentionally stops an active Goal and permission
// selection is an independent composer axis.
// A change that does not widen the preset invalidates the active turn so an
// older approval never authorizes work the new snapshot forbids. A widening
// revokes nothing the turn holds, so the turn keeps running. Avoid the idle
// Cancel path: it stops an active Goal, and the preset is a separate axis.
upgrade := permissionPresetRank(mode) > permissionPresetRank(previousMode)
turnID, cancelled := "", false
if c.Running() {
if !upgrade && c.Running() {
turnID, cancelled = c.cancelTurnLocked()
}
c.promptResolveMu.Unlock()
+71
View File
@@ -0,0 +1,71 @@
package control
import (
"context"
"testing"
"time"
"reasonix/internal/agent"
"reasonix/internal/provider"
)
type presetTurnRunner struct {
session *agent.Session
release chan struct{}
}
func (r presetTurnRunner) Run(ctx context.Context, input string) error {
r.session.Add(provider.Message{Role: provider.RoleUser, Content: input})
select {
case <-r.release:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func TestPermissionPresetChangeCancelsRunningTurnOnlyWhenNotWidening(t *testing.T) {
cases := []struct {
name string
from, to string
wantCancel bool
}{
{"workspace to full access", ToolApprovalWorkspaceWrite, ToolApprovalDangerFullAccess, false},
{"read-only to workspace", ToolApprovalReadOnly, ToolApprovalWorkspaceWrite, false},
{"full access to workspace", ToolApprovalDangerFullAccess, ToolApprovalWorkspaceWrite, true},
{"workspace to read-only", ToolApprovalWorkspaceWrite, ToolApprovalReadOnly, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
release := make(chan struct{})
c := newOwnedTestController(t, Options{Runner: presetTurnRunner{session: agent.NewSession("sys"), release: release}})
c.SetToolApprovalMode(tc.from)
done := make(chan error, 1)
go func() { done <- c.RunTurn(context.Background(), "work") }()
waitForRunning(t, c)
c.SetToolApprovalMode(tc.to)
if got := c.ToolApprovalMode(); got != tc.to {
t.Fatalf("mode = %q, want %q", got, tc.to)
}
if tc.wantCancel {
waitIdle(t, c)
} else {
select {
case err := <-done:
t.Fatalf("widening the preset ended the running turn: %v", err)
case <-time.After(200 * time.Millisecond):
}
if !c.Running() {
t.Fatal("widening the preset stopped the running turn")
}
}
close(release)
select {
case <-done:
case <-time.After(30 * time.Second):
t.Fatal("turn did not finish")
}
})
}
}
@@ -0,0 +1,46 @@
package control
import (
"context"
"testing"
"reasonix/internal/event"
"reasonix/internal/provider"
)
// The approved-plan execution turn runs in the planning turn's context, which
// carries the user message identity the controller reserved for that turn.
func TestApprovedPlanExecutionMintsItsOwnUserMessageID(t *testing.T) {
prov := &scriptedTurns{turns: planThenExecuteTurns("Plan:\n1. Change it", "Done.")}
ag := newPlanTestAgent(prov)
approvalID := make(chan string, 1)
c := newOwnedTestController(t, Options{
Runner: ag,
Executor: ag,
Sink: event.FuncSink(func(e event.Event) {
if e.Kind == event.ApprovalRequest {
approvalID <- e.Approval.ID
}
}),
})
c.SetPlanMode(true)
go func() { c.Approve(<-approvalID, true, false, false) }()
if err := c.runTurnWithRaw(context.Background(), "change it", "change it"); err != nil {
t.Fatalf("runTurnWithRaw: %v", err)
}
users := 0
seen := map[string]bool{}
for _, message := range ag.Session().Messages {
if message.Role == provider.RoleUser {
users++
}
if seen[message.ID] {
t.Fatalf("message id %q appears twice (second as %s/%s)", message.ID, message.Role, message.Origin)
}
seen[message.ID] = true
}
if users < 2 {
t.Fatalf("user messages = %d, want the request and the approved-plan continuation", users)
}
}
+5 -1
View File
@@ -52,6 +52,10 @@ type SubmissionAttachment struct {
// the caller must query the receipt before deciding whether a retry may run.
var ErrSubmissionNotAccepted = errors.New("submission not accepted")
// ErrSubmissionIdentityUnavailable marks a runtime with no durable store to
// record an identified submission in, such as a native legacy transcript.
var ErrSubmissionIdentityUnavailable = errors.New("durable submission identity unavailable")
type submissionIdentityState struct {
mu sync.Mutex
pending atomic.Pointer[pendingSubmissionAdmission]
@@ -279,7 +283,7 @@ func (c *Controller) acceptPreparedSubmission(ctx context.Context, candidate *Pr
store := c.sessionEventStore()
if req.ID != "" {
if store == nil {
return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, errors.New("durable submission identity unavailable"))
return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, ErrSubmissionIdentityUnavailable)
}
if c.Running() {
return session.SubmissionReceipt{}, errors.Join(ErrSubmissionNotAccepted, ErrTurnRunning)
@@ -0,0 +1,174 @@
package control
import (
"context"
"encoding/json"
"errors"
"fmt"
"path/filepath"
"testing"
"time"
"reasonix/internal/agent"
"reasonix/internal/agent/testutil"
"reasonix/internal/event"
"reasonix/internal/provider"
"reasonix/internal/session"
"reasonix/internal/tool"
"reasonix/internal/transcript"
)
type followTally struct {
polls, changes, maxBatch int
resets, gaps, changeResets int
}
// desktopFollower applies transcriptFollowClient's acceptance rules: a
// response reset, a change reset or a skipped revision each cost a baseline.
type desktopFollower struct {
c *Controller
subscription string
revision uint64
tally followTally
}
func (f *desktopFollower) baseline(ctx context.Context) error {
response, err := f.c.TranscriptFollow(ctx, transcript.FollowRequest{})
if err == nil && response.Snapshot == nil {
err = errors.New("baseline without snapshot")
}
if err != nil {
return err
}
f.subscription, f.revision = response.Subscription, response.Snapshot.ProjectionRevision
return nil
}
// run polls until done is closed and the queue is drained; work stands in for
// the renderer applying one delivered batch before it asks for the next.
func (f *desktopFollower) run(ctx context.Context, done <-chan struct{}, work time.Duration) error {
finished := false
for {
select {
case <-done:
finished = true
default:
}
pollCtx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
response, err := f.c.TranscriptFollow(pollCtx, transcript.FollowRequest{Subscription: f.subscription, AfterRevision: f.revision})
cancel()
if err != nil && !errors.Is(err, context.DeadlineExceeded) {
return err
}
f.tally.polls++
f.tally.changes += len(response.Changes)
f.tally.maxBatch = max(f.tally.maxBatch, len(response.Changes))
broken := response.ResetRequired
if broken {
f.tally.resets++
}
for _, change := range response.Changes {
if broken || change.Revision <= f.revision {
continue
}
switch {
case change.ResetRequired:
f.tally.changeResets++
broken = true
case change.Revision != f.revision+1:
f.tally.gaps++
broken = true
default:
f.revision = change.Revision
}
}
if broken {
_, _ = f.c.TranscriptFollow(context.Background(), transcript.FollowRequest{Subscription: f.subscription, Close: true})
if err := f.baseline(ctx); err != nil {
return err
}
continue
}
if finished && len(response.Changes) == 0 {
_, _ = f.c.TranscriptFollow(context.Background(), transcript.FollowRequest{Subscription: f.subscription, Close: true})
return nil
}
time.Sleep(work)
}
}
// shellFloodTool writes output the way a verbose build does: many short
// lines, each reaching the agent as its own progress chunk.
type shellFloodTool struct{ lines int }
func (shellFloodTool) Name() string { return "flood" }
func (shellFloodTool) Description() string { return "prints build output" }
func (shellFloodTool) Schema() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
func (shellFloodTool) ReadOnly() bool { return true }
func (f shellFloodTool) Execute(ctx context.Context, _ json.RawMessage) (string, error) {
emit, _ := tool.ProgressFrom(ctx)
for i := range f.lines {
if emit != nil {
emit(fmt.Sprintf("compiling unit %d\n", i))
}
if i%2 == 1 {
time.Sleep(time.Millisecond)
}
}
return "ok", nil
}
func TestTranscriptFollowSteadyStreamingNeverResetsFollower(t *testing.T) {
chunks := make([]provider.Chunk, 0, 3001)
for range 1500 {
chunks = append(chunks, provider.Chunk{Type: provider.ChunkReasoning, Text: "think "})
}
for range 1500 {
chunks = append(chunks, provider.Chunk{Type: provider.ChunkText, Text: "word "})
}
chunks = append(chunks, provider.Chunk{Type: provider.ChunkDone})
flood := []provider.ToolCall{{ID: "call-flood", Name: "flood", Arguments: "{}"}}
cases := map[string][]testutil.Turn{
"reasoning and text": {{Chunks: chunks}},
"shell output": {{ToolCalls: flood}, {Text: "built"}},
}
for name, turns := range cases {
t.Run(name, func(t *testing.T) {
service, err := session.NewService("desktop", session.NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions")))
if err != nil {
t.Fatal(err)
}
runtime, err := service.Create(t.Context(), session.CreateOptions{SessionID: "transcript-follow"})
if err != nil {
t.Fatal(err)
}
registry := tool.NewRegistry()
registry.Add(shellFloodTool{lines: 3000})
executor := agent.New(testutil.NewMock("test", turns...), registry, agent.NewSession("system"), agent.Options{}, event.Discard)
c := newOwnedTestController(t, Options{Runner: executor, Executor: executor, Sink: event.Discard,
SessionService: service, SessionRuntime: runtime, ExclusiveSession: true})
follower := &desktopFollower{c: c}
if err := follower.baseline(t.Context()); err != nil {
t.Fatal(err)
}
done, result := make(chan struct{}), make(chan error, 1)
go func() { result <- follower.run(t.Context(), done, 400*time.Millisecond) }()
started := time.Now()
if err := c.RunTurn(t.Context(), "question"); err != nil {
t.Fatal(err)
}
elapsed := time.Since(started)
close(done)
if err := <-result; err != nil {
t.Fatal(err)
}
tally := follower.tally
t.Logf("turn=%v polls=%d changes=%d maxBatch=%d resets=%d gaps=%d changeResets=%d",
elapsed, tally.polls, tally.changes, tally.maxBatch, tally.resets, tally.gaps, tally.changeResets)
if tally.resets+tally.gaps+tally.changeResets != 0 {
t.Fatalf("steady output forced %d follower baselines: %+v", tally.resets+tally.gaps+tally.changeResets, tally)
}
})
}
}
+4
View File
@@ -54,6 +54,10 @@ func (o *turnOrchestrator) runComposedSyntheticTurn(ctx context.Context, text st
c := o.c
ctx = agent.WithRawUserInput(ctx, text)
ctx = withTurnInputOrigin(ctx, true)
// ctx may carry the identity already spent on the turn's own user message.
if c.executor != nil {
ctx = agent.WithUserMessageIdentity(ctx, c.executor.Session(), agent.NewMessageID())
}
ctx = c.withTurnContext(ctx, false)
ctx = c.withPlannerTurnMetadata(ctx, text, true, c.messageCount())
return c.runModelTurn(ctx, c.ComposeSynthetic(text))
+48 -27
View File
@@ -21,7 +21,8 @@ const coalesceMaxBytes = 16 << 10
const DefaultStreamDeltaWindow = 16 * time.Millisecond
// Coalesce wraps inner so bursts of consecutive streaming deltas — Text or
// Reasoning events carrying nothing but a Text payload — merge into one event.
// Reasoning events carrying nothing but a Text payload, or ToolProgress events
// carrying nothing but one tool's Output — merge into one event.
// The first delta of a burst forwards immediately (time-to-first-token is
// unchanged); later deltas buffer at most window, flushing earlier on any
// other event (total order preserved), a kind switch, or coalesceMaxBytes.
@@ -43,10 +44,7 @@ type coalescer struct {
// called under mu. A single drainer forwards FIFO, so a sink that
// synchronously re-enters Emit enqueues and returns instead of deadlocking.
mu sync.Mutex
kind Kind
source string
messageID string
attemptID string
key deltaKey
buf strings.Builder
pending bool
timer *time.Timer
@@ -64,19 +62,44 @@ type coalescedEvent struct {
var _ OptionalSinkCapabilities = (*coalescer)(nil)
var _ CheckedSink = (*coalescer)(nil)
// isStreamDelta reports whether e is a pure streaming delta: merging is only
// safe when no other field carries meaning. The zero-probe comparison keeps
// this true by construction as Event grows fields.
func isStreamDelta(e Event) bool {
if (e.Kind != Text && e.Kind != Reasoning) || e.Text == "" {
return false
}
// deltaKey is the identity a buffered burst merges under; any change flushes.
type deltaKey struct {
kind Kind
source, messageID, attemptID string
toolID string
}
// streamDelta reports whether e is a pure streaming delta, with its merge key
// and payload: merging is only safe when no other field carries meaning. The
// zero-probe comparison keeps this true by construction as Event grows fields.
func streamDelta(e Event) (deltaKey, string, bool) {
key := deltaKey{kind: e.Kind, source: e.Source, messageID: e.MessageID, attemptID: e.AttemptID}
probe := e
probe.Text = ""
probe.Source = ""
probe.MessageID = ""
probe.AttemptID = ""
return reflect.DeepEqual(probe, Event{Kind: e.Kind})
probe.Source, probe.MessageID, probe.AttemptID = "", "", ""
payload := e.Text
switch e.Kind {
case Text, Reasoning:
probe.Text = ""
case ToolProgress:
key.toolID, payload = e.Tool.ID, e.Tool.Output
probe.Tool.ID, probe.Tool.Output = "", ""
default:
return deltaKey{}, "", false
}
if payload == "" || (e.Kind == ToolProgress && key.toolID == "") || !reflect.DeepEqual(probe, Event{Kind: e.Kind}) {
return deltaKey{}, "", false
}
return key, payload, true
}
func (k deltaKey) event(payload string) Event {
e := Event{Kind: k.kind, Source: k.source, MessageID: k.messageID, AttemptID: k.attemptID}
if k.kind == ToolProgress {
e.Tool = Tool{ID: k.toolID, Output: payload}
} else {
e.Text = payload
}
return e
}
func (c *coalescer) Emit(e Event) {
@@ -97,14 +120,15 @@ func (c *coalescer) enqueue(e Event, checked bool) error {
if checked {
done = make(chan error, 1)
}
key, payload, delta := streamDelta(e)
c.mu.Lock()
if checked && isStreamDelta(e) {
if checked && delta {
c.enqueueFlushLocked()
c.queue = append(c.queue, coalescedEvent{event: e, done: done})
c.drainAndUnlock()
return <-done
}
if !isStreamDelta(e) {
if !delta {
c.enqueueFlushLocked()
c.queue = append(c.queue, coalescedEvent{event: e, done: done})
c.drainAndUnlock()
@@ -113,7 +137,7 @@ func (c *coalescer) enqueue(e Event, checked bool) error {
}
return nil
}
if c.pending && (c.kind != e.Kind || c.source != e.Source || c.messageID != e.MessageID || c.attemptID != e.AttemptID) {
if c.pending && c.key != key {
c.enqueueFlushLocked()
}
if !c.pending && time.Since(c.lastForward) >= c.window {
@@ -127,17 +151,14 @@ func (c *coalescer) enqueue(e Event, checked bool) error {
}
if !c.pending {
c.pending = true
c.kind = e.Kind
c.source = e.Source
c.messageID = e.MessageID
c.attemptID = e.AttemptID
c.key = key
if c.timer == nil {
c.timer = time.AfterFunc(c.window, c.flush)
} else {
c.timer.Reset(c.window)
}
}
c.buf.WriteString(e.Text)
c.buf.WriteString(payload)
if c.buf.Len() >= coalesceMaxBytes {
c.enqueueFlushLocked()
}
@@ -160,10 +181,10 @@ func (c *coalescer) enqueueFlushLocked() {
return
}
c.timer.Stop()
c.queue = append(c.queue, coalescedEvent{event: Event{Kind: c.kind, Text: c.buf.String(), Source: c.source, MessageID: c.messageID, AttemptID: c.attemptID}})
c.queue = append(c.queue, coalescedEvent{event: c.key.event(c.buf.String())})
c.buf.Reset()
c.pending = false
c.source = ""
c.key = deltaKey{}
c.lastForward = time.Now()
}
+31
View File
@@ -259,6 +259,37 @@ func TestCoalesceCapabilityCannotOvertakeActiveDrainer(t *testing.T) {
}
}
func TestCoalesceMergesToolOutputPerTool(t *testing.T) {
inner := &coalesceRecordSink{}
c := Coalesce(inner, time.Hour)
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "a", Output: "1\n"}}) // leading edge
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "a", Output: "2\n"}})
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "a", Output: "3\n"}})
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "b", Output: "x\n"}}) // other tool: flush + buffer
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "b", Verifying: true}})
c.Emit(Event{Kind: ToolProgress, Tool: Tool{ID: "c", Name: SubagentProgressStatusName, Output: "running"}})
c.Emit(Event{Kind: ToolResult, Tool: Tool{ID: "a", Output: "done"}})
got := inner.snapshot()
want := []Event{
{Kind: ToolProgress, Tool: Tool{ID: "a", Output: "1\n"}},
{Kind: ToolProgress, Tool: Tool{ID: "a", Output: "2\n3\n"}},
{Kind: ToolProgress, Tool: Tool{ID: "b", Output: "x\n"}},
{Kind: ToolProgress, Tool: Tool{ID: "b", Verifying: true}},
{Kind: ToolProgress, Tool: Tool{ID: "c", Name: SubagentProgressStatusName, Output: "running"}},
{Kind: ToolResult, Tool: Tool{ID: "a", Output: "done"}},
}
if len(got) != len(want) {
t.Fatalf("got %d events, want %d: %+v", len(got), len(want), got)
}
for i := range want {
if got[i].Kind != want[i].Kind || got[i].Tool.ID != want[i].Tool.ID || got[i].Tool.Output != want[i].Tool.Output ||
got[i].Tool.Verifying != want[i].Tool.Verifying || got[i].Tool.Name != want[i].Tool.Name {
t.Fatalf("event %d = %+v, want %+v", i, got[i], want[i])
}
}
}
func TestCoalesceNonPureDeltaPassesThrough(t *testing.T) {
inner := &coalesceRecordSink{}
c := Coalesce(inner, time.Hour)
+42
View File
@@ -0,0 +1,42 @@
package pathidentity
import (
"errors"
"strings"
"syscall"
"golang.org/x/text/unicode/norm"
)
// darwinVolume answers the two questions the darwin identity key asks of the
// volume holding a directory. It is platform-neutral so the rules are testable
// on every host; key_darwin.go binds it to statfs(2) and pathconf(2).
type darwinVolume struct {
fsType func(dir string) (string, error)
caseSensitive func(dir string) (int, error)
}
func darwinIdentityKeyBy(path, parent string, volume darwinVolume) (string, error) {
// A volume whose type cannot be read is not known to normalize names, so
// the path is kept as spelled rather than failing the whole lookup.
if fsType, err := volume.fsType(parent); err == nil && (fsType == "apfs" || fsType == "hfs") {
path = norm.NFD.String(path)
}
caseSensitive, err := volume.caseSensitive(parent)
if err != nil {
// FUSE volumes (sshfs over macFUSE) do not implement
// _PC_CASE_SENSITIVE; an undeclared answer keeps names distinct.
if !pathconfUndeclared(err) {
return "", err
}
caseSensitive = 1
}
if caseSensitive == 0 {
path = strings.ToLower(path)
}
return path, nil
}
func pathconfUndeclared(err error) bool {
return errors.Is(err, syscall.EINVAL) || errors.Is(err, syscall.ENOTSUP) || errors.Is(err, syscall.EOPNOTSUPP)
}
@@ -0,0 +1,49 @@
package pathidentity
import (
"errors"
"fmt"
"syscall"
"testing"
)
func TestDarwinIdentityKeyToleratesVolumesThatDoNotAnswer(t *testing.T) {
const path = "/Volumes/remote/Proj/Café.go"
fixed := func(value string, err error) func(string) (string, error) {
return func(string) (string, error) { return value, err }
}
pathconf := func(value int, err error) func(string) (int, error) {
return func(string) (int, error) { return value, err }
}
cases := []struct {
name string
volume darwinVolume
want string
}{
{"fuse pathconf EINVAL", darwinVolume{fixed("macfuse", nil), pathconf(-1, syscall.EINVAL)}, path},
{"fuse pathconf ENOTSUP", darwinVolume{fixed("macfuse", nil), pathconf(-1, fmt.Errorf("pathconf: %w", syscall.ENOTSUP))}, path},
{"statfs failure", darwinVolume{fixed("", syscall.EIO), pathconf(1, nil)}, path},
{"case-insensitive apfs", darwinVolume{fixed("apfs", nil), pathconf(0, nil)}, "/volumes/remote/proj/café.go"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got, err := darwinIdentityKeyBy(path, "/Volumes/remote", tc.volume)
if err != nil {
t.Fatalf("identity key failed: %v", err)
}
if got != tc.want {
t.Fatalf("identity key = %q, want %q", got, tc.want)
}
})
}
}
func TestDarwinIdentityKeyKeepsOtherPathconfFailures(t *testing.T) {
volume := darwinVolume{
fsType: func(string) (string, error) { return "apfs", nil },
caseSensitive: func(string) (int, error) { return -1, syscall.EACCES },
}
if _, err := darwinIdentityKeyBy("/p", "/", volume); !errors.Is(err, syscall.EACCES) {
t.Fatalf("error = %v, want EACCES", err)
}
}
+14 -17
View File
@@ -9,32 +9,29 @@ import (
"syscall"
"golang.org/x/sys/unix"
"golang.org/x/text/unicode/norm"
)
const pathconfCaseSensitive = 11
var systemVolume = darwinVolume{
fsType: func(dir string) (string, error) {
var stat unix.Statfs_t
if err := unix.Statfs(dir, &stat); err != nil {
return "", err
}
return strings.TrimRight(string(stat.Fstypename[:]), "\x00"), nil
},
caseSensitive: func(dir string) (int, error) {
return syscall.Pathconf(dir, pathconfCaseSensitive)
},
}
func platformIdentityKey(path string) (string, error) {
parent, err := closestExistingDirectory(path)
if err != nil {
return "", err
}
var stat unix.Statfs_t
if err := unix.Statfs(parent, &stat); err != nil {
return "", err
}
fsType := strings.TrimRight(string(stat.Fstypename[:]), "\x00")
if fsType == "apfs" || fsType == "hfs" {
path = norm.NFD.String(path)
}
caseSensitive, err := syscall.Pathconf(parent, pathconfCaseSensitive)
if err != nil {
return "", err
}
if caseSensitive == 0 {
path = strings.ToLower(path)
}
return path, nil
return darwinIdentityKeyBy(path, parent, systemVolume)
}
func closestExistingDirectory(path string) (string, error) {
+1 -1
View File
@@ -182,7 +182,7 @@ func TestCacheLoadQuarantinesMalformedToolSchema(t *testing.T) {
if len(got.Tools) != 1 || got.Tools[0].Name != "do_thing" {
t.Fatalf("cached tools = %+v, want only valid do_thing", got.Tools)
}
if schema := string(got.Tools[0].Schema); schema != `{"properties":{},"type":"object"}` {
if schema := string(got.Tools[0].Schema); schema != `{"properties":{},"required":[],"type":"object"}` {
t.Fatalf("valid cached schema = %s", schema)
}
}
+1 -1
View File
@@ -46,7 +46,7 @@ func TestCanonicalizeSchemaSortsKeys(t *testing.T) {
result := canonicalizeSchema(schema)
// json.Marshal sorts map keys, so verify the JSON string directly.
s := string(result)
if s != `{"a":2,"m":3,"properties":{},"type":"object","z":1}` {
if s != `{"a":2,"m":3,"properties":{},"required":[],"type":"object","z":1}` {
t.Errorf("keys not sorted, got: %s", s)
}
}
+3 -3
View File
@@ -443,7 +443,7 @@ func TestClientListToolsQuarantinesMalformedSchema(t *testing.T) {
if len(tools) != 1 || tools[0].Name() != "mcp__yakit__echo" {
t.Fatalf("tools = %v, want only mcp__yakit__echo", names(tools))
}
if got := string(tools[0].Schema()); got != `{"properties":{"msg":{"type":"string"}},"type":"object"}` {
if got := string(tools[0].Schema()); got != `{"properties":{"msg":{"type":"string"}},"required":[],"type":"object"}` {
t.Fatalf("valid sibling schema changed: %s", got)
}
if len(c.toolCatalog.infos) != 2 {
@@ -478,7 +478,7 @@ func TestClientListToolsQuarantinesNonObjectRootSchemas(t *testing.T) {
if len(tools) != 2 || tools[0].Name() != "mcp__srv__echo" || tools[1].Name() != "mcp__srv__no_args" {
t.Fatalf("tools = %v, want echo and normalized no_args", names(tools))
}
if got := string(tools[1].Schema()); got != `{"properties":{},"type":"object"}` {
if got := string(tools[1].Schema()); got != `{"properties":{},"required":[],"type":"object"}` {
t.Fatalf("no_args schema = %s, want normalized empty object schema", got)
}
if len(c.toolCatalog.infos) != 4 {
@@ -514,7 +514,7 @@ func TestClientListToolsValidatesAfterCompatibilityNormalization(t *testing.T) {
if len(tools) != 1 {
t.Fatalf("tools = %v, want normalized legacy tool", names(tools))
}
if got := string(tools[0].Schema()); got != `{"properties":{"query":{"type":"string"}},"type":"object"}` {
if got := string(tools[0].Schema()); got != `{"properties":{"query":{"type":"string"}},"required":[],"type":"object"}` {
t.Fatalf("normalized schema = %s", got)
}
}
+15 -1
View File
@@ -82,6 +82,18 @@ type OpenOptions struct {
ResumeKey string
}
// WALSizeLimit is the size a disk projection's WAL is truncated back to after
// a checkpoint.
const WALSizeLimit = 4 << 20
// CheckpointBeforeClose folds the WAL into the database and truncates it. It is
// best-effort: another process's reader leaves the WAL for its next checkpoint.
func CheckpointBeforeClose(ctx context.Context, db *sql.DB) {
if db != nil {
_, _ = db.ExecContext(ctx, `PRAGMA wal_checkpoint(TRUNCATE)`)
}
}
type Handle struct {
DB *sql.DB
Status Status
@@ -207,8 +219,10 @@ func open(ctx context.Context, opts OpenOptions, mode Mode, checkIntegrity bool)
opts.Now().UnixNano(), memoryDatabaseSequence.Add(1))
} else {
var err error
// journal_size_limit is per connection, so it rides the DSN to reach
// every pooled one; without it a checkpointed WAL keeps its peak size.
dsn, err = sqliteuri.Disk(opts.Path, url.Values{
"_pragma": {"busy_timeout(150)", "foreign_keys(1)"},
"_pragma": {"busy_timeout(150)", "foreign_keys(1)", fmt.Sprintf("journal_size_limit(%d)", WALSizeLimit)},
})
if err != nil {
return nil, err
@@ -352,3 +352,79 @@ func TestRebuildHoldsExclusiveLifecycleLock(t *testing.T) {
}
release()
}
func TestDiskOpenLimitsWALOnEveryPooledConnection(t *testing.T) {
ctx := context.Background()
handle, err := Open(ctx, OpenOptions{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), Migrations: testMigrations(), MaxOpenConns: 4})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = handle.DB.Close() })
conns := make([]*sql.Conn, 0, 4)
for range 4 {
conn, err := handle.DB.Conn(ctx)
if err != nil {
t.Fatal(err)
}
conns = append(conns, conn)
}
for i, conn := range conns {
var limit int64
if err := conn.QueryRowContext(ctx, `PRAGMA journal_size_limit`).Scan(&limit); err != nil || limit != WALSizeLimit {
t.Fatalf("connection %d journal_size_limit=%d err=%v", i, limit, err)
}
}
for _, conn := range conns {
_ = conn.Close()
}
}
// A reader pinning an old snapshot lets the WAL grow past the limit; once it
// is released, a checkpoint must bring the file back under it (#10714).
func TestCheckpointedWALShrinksToLimit(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "catalog.sqlite")
handle, err := Open(ctx, OpenOptions{Path: path, Migrations: testMigrations(), MaxOpenConns: 4})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = handle.DB.Close() })
reader, err := handle.DB.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
t.Fatal(err)
}
if err := reader.QueryRowContext(ctx, `SELECT count(*) FROM values_table`).Scan(new(int)); err != nil {
t.Fatal(err)
}
for range 4 {
if _, err := handle.DB.ExecContext(ctx, `INSERT INTO values_table(value)
WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM n WHERE x < 1024)
SELECT hex(randomblob(1024)) FROM n`); err != nil {
t.Fatal(err)
}
}
walSize := func() int64 {
info, err := os.Stat(path + "-wal")
if err != nil {
t.Fatal(err)
}
return info.Size()
}
if size := walSize(); size <= WALSizeLimit {
t.Fatalf("fixture WAL=%d did not exceed the limit", size)
}
_ = reader.Rollback()
if _, err := handle.DB.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE)`); err != nil {
t.Fatal(err)
}
if _, err := handle.DB.ExecContext(ctx, `INSERT INTO values_table(value) VALUES('after')`); err != nil {
t.Fatal(err)
}
if size := walSize(); size > WALSizeLimit {
t.Fatalf("WAL after checkpoint=%d, limit %d", size, WALSizeLimit)
}
CheckpointBeforeClose(ctx, handle.DB)
if size := walSize(); size != 0 {
t.Fatalf("WAL after close checkpoint=%d", size)
}
}
+2 -2
View File
@@ -2119,7 +2119,7 @@ func TestBuildRequestContentNullForAssistantToolCalls(t *testing.T) {
if !strings.Contains(s, `"content":"all done"`) {
t.Errorf("text assistant turn should keep its string content: %s", s)
}
if !strings.Contains(s, `"parameters":{"properties":{},"type":"object"}`) {
if !strings.Contains(s, `"parameters":{"properties":{},"required":[],"type":"object"}`) {
t.Errorf("no-param tool should serialize a strict empty-object schema: %s", s)
}
}
@@ -2176,7 +2176,7 @@ func TestBuildRequestDefaultsEmptyToolParameters(t *testing.T) {
if _, ok := fn["description"]; ok {
t.Fatalf("empty description should be omitted: %s", body)
}
if got, want := string(fn["parameters"]), `{"properties":{},"type":"object"}`; got != want {
if got, want := string(fn["parameters"]), `{"properties":{},"required":[],"type":"object"}`; got != want {
t.Fatalf("nil parameters should default to %s, got %s in %s", want, got, body)
}
}
+7 -2
View File
@@ -13,7 +13,7 @@ func CanonicalizeSchema(raw json.RawMessage) json.RawMessage {
// schema. An empty json.RawMessage makes json.Marshal of the enclosing
// request fail ("unexpected end of JSON input") and bricks the whole
// provider; emit a strict OpenAI-compatible empty-object schema instead.
return json.RawMessage(`{"properties":{},"type":"object"}`)
return json.RawMessage(`{"properties":{},"required":[],"type":"object"}`)
}
var v any
if err := json.Unmarshal(raw, &v); err != nil {
@@ -23,7 +23,7 @@ func CanonicalizeSchema(raw json.RawMessage) json.RawMessage {
// A nil RawMessage persists as JSON null in the MCP schema cache. Treat
// both forms as the same no-argument schema so old cache entries remain
// usable and never reach a strict provider as parameters: null.
return json.RawMessage(`{"properties":{},"type":"object"}`)
return json.RawMessage(`{"properties":{},"required":[],"type":"object"}`)
}
canon := canonicalizeSchemaValue(v)
ensureRootObjectProperties(canon)
@@ -53,6 +53,11 @@ func ensureRootObjectProperties(v any) {
if _, ok := m["properties"]; !ok {
m["properties"] = map[string]any{}
}
// Relays that re-serialize schemas into typed structs emit an omitted
// required as null, which strict upstreams reject as "not of type array".
if _, ok := m["required"]; !ok {
m["required"] = []any{}
}
}
func canonicalizeSchemaValue(v any) any {
+18 -6
View File
@@ -29,17 +29,29 @@ func TestCanonicalizeSchemaAddsEmptyPropertiesForNoArgumentObject(t *testing.T)
json.RawMessage(`{"type":"object"}`),
} {
got := string(CanonicalizeSchema(raw))
want := `{"properties":{},"type":"object"}`
want := `{"properties":{},"required":[],"type":"object"}`
if got != want {
t.Fatalf("CanonicalizeSchema(%s) = %s, want %s", string(raw), got, want)
}
}
}
func TestCanonicalizeSchemaRootObjectCarriesRequiredArray(t *testing.T) {
for _, tc := range []struct{ raw, want string }{
{`{"type":"object","additionalProperties":false,"properties":{}}`, `{"additionalProperties":false,"properties":{},"required":[],"type":"object"}`},
{`{"type":"object","properties":{},"required":null}`, `{"properties":{},"required":[],"type":"object"}`},
{`{"type":"object","properties":{"q":{"type":"object","properties":{}}}}`, `{"properties":{"q":{"properties":{},"type":"object"}},"required":[],"type":"object"}`},
} {
if got := string(CanonicalizeSchema(json.RawMessage(tc.raw))); got != tc.want {
t.Fatalf("CanonicalizeSchema(%s) = %s, want %s", tc.raw, got, tc.want)
}
}
}
func TestCanonicalizeSchemaAddsMissingRootType(t *testing.T) {
for _, tc := range []struct{ raw, want string }{
{`{}`, `{"properties":{},"type":"object"}`},
{`{"properties":{"q":{"type":"string"}}}`, `{"properties":{"q":{"type":"string"}},"type":"object"}`},
{`{}`, `{"properties":{},"required":[],"type":"object"}`},
{`{"properties":{"q":{"type":"string"}}}`, `{"properties":{"q":{"type":"string"}},"required":[],"type":"object"}`},
// Explicit non-object root types are preserved verbatim; validation
// quarantines them instead of silently rewriting declared semantics.
{`{"type":"string"}`, `{"type":"string"}`},
@@ -78,7 +90,7 @@ func TestCanonicalizeSchemaDependentRequired(t *testing.T) {
}`)
got := string(CanonicalizeSchema(raw))
want := `{"dependentRequired":{"cc":["billing_address","name"]},"properties":{},"type":"object"}`
want := `{"dependentRequired":{"cc":["billing_address","name"]},"properties":{},"required":[],"type":"object"}`
if got != want {
t.Fatalf("CanonicalizeSchema() = %s, want %s", got, want)
}
@@ -97,7 +109,7 @@ func TestCanonicalizeSchemaPreservesDependentRequiredPropertyName(t *testing.T)
}`)
got := string(CanonicalizeSchema(raw))
want := `{"dependentRequired":{"dependentRequired":["x"]},"properties":{"dependentRequired":{"type":"string"},"x":{"type":"string"}},"type":"object"}`
want := `{"dependentRequired":{"dependentRequired":["x"]},"properties":{"dependentRequired":{"type":"string"},"x":{"type":"string"}},"required":[],"type":"object"}`
if got != want {
t.Fatalf("CanonicalizeSchema() = %s, want %s", got, want)
}
@@ -173,7 +185,7 @@ func TestNormalizeLegacyTupleItemsPreservesModernArrayItems(t *testing.T) {
func TestCanonicalizeSchemaPreservesLegacyTupleItems(t *testing.T) {
raw := json.RawMessage(`{"type":"object","properties":{"pair":{"type":"array","items":[{"type":"string"},{"type":"number"}],"additionalItems":false}}}`)
want := `{"properties":{"pair":{"additionalItems":false,"items":[{"type":"string"},{"type":"number"}],"type":"array"}},"type":"object"}`
want := `{"properties":{"pair":{"additionalItems":false,"items":[{"type":"string"},{"type":"number"}],"type":"array"}},"required":[],"type":"object"}`
if got := string(CanonicalizeSchema(raw)); got != want {
t.Fatalf("CanonicalizeSchema() = %s, want provider-neutral bytes %s", got, want)
}
+16 -8
View File
@@ -20,9 +20,9 @@ var supersededAppUpdateAfterBackupArchive = func(string) {}
// ArchiveSupersededPendingAppBundleUpdate retires a legacy macOS transaction
// only after a healthy desktop is already running from the transaction's exact
// target bundle. It is intentionally limited to the two unrecoverable legacy
// shapes: the rollback backup identity was never recorded, or the recorded
// backup no longer exists. A surviving backup is content-bound and moved aside;
// target bundle, and only for records the health commit cannot settle: no
// recorded backup identity, a recorded backup that is gone, or a target already
// reached whose record no longer verifies. A surviving backup is moved aside;
// the original transaction is archived under Reasonix repair state. Neither is
// deleted, so support can still inspect or manually recover the old bundle.
func ArchiveSupersededPendingAppBundleUpdate(runningVersion string) (bool, error) {
@@ -142,10 +142,10 @@ func validateSupersededPendingAppBundleUpdate(tx *UpdateTransaction, runningVers
if err := validateUpdateTransaction(tx); err != nil {
return false, fmt.Errorf("archive superseded app update: invalid transaction: %w", err)
}
// A normal update prepares the transaction before this process shuts down;
// its backup is intentionally absent until the detached helper performs the
// swap. Never mistake that fresh handoff for a stale missing-backup record.
if tx.HandoffOwnerPID == os.Getpid() {
// A fresh handoff has no backup until the detached helper swaps the bundle.
// Its owner is this process under Wails and the Electron parent in host mode
// (desktop updateHandoffOwnerPID); either one means the record is live.
if tx.HandoffOwnerPID > 0 && (tx.HandoffOwnerPID == os.Getpid() || tx.HandoffOwnerPID == os.Getppid()) {
return false, nil
}
running := canonicalSemver(runningVersion)
@@ -165,7 +165,15 @@ func validateSupersededPendingAppBundleUpdate(tx *UpdateTransaction, runningVers
} else if err != nil {
return false, fmt.Errorf("archive superseded app update: inspect rollback backup: %w", err)
}
return false, nil
if semver.Compare(running, to) < 0 {
return false, nil
}
// The health commit owns a record it can still prove. One it cannot, such as
// a backup Finder wrote .DS_Store into, would pin every later update.
if semver.Compare(running, to) == 0 && pendingUpdateInstalledForHealth(tx) {
return false, nil
}
return true, nil
}
func archiveSupersededAppBundleBackup(tx *UpdateTransaction, transactionID string) (string, string, error) {
+75 -10
View File
@@ -136,8 +136,30 @@ func TestArchiveSupersededPendingAppBundleUpdateHealsMissingBackup(t *testing.T)
}
}
// recordHandoff gives tx the metadata PrepareAppBundleUpdateHandoff writes, so
// ReadPendingUpdate accepts it.
func recordHandoff(t *testing.T, tx *UpdateTransaction, ownerPID int, appTreeID string) {
t.Helper()
tx.HandoffOwnerPID = ownerPID
tx.HandoffStagingPath = filepath.Join(os.TempDir(), "reasonix-mac-update-legacy-test")
tx.HandoffAppPath = filepath.Join(tx.HandoffStagingPath, "Reasonix.app")
tx.HandoffAppTreeID = appTreeID
tx.HandoffStagingTreeID = strings.Repeat("c", 64)
if err := overwritePendingUpdateForTest(tx); err != nil {
t.Fatal(err)
}
if _, err := ReadPendingUpdate(); err != nil {
t.Fatalf("recorded handoff is not readable: %v", err)
}
}
func TestArchiveSupersededPendingAppBundleUpdateLeavesModernRollbackForHealthCommit(t *testing.T) {
prepareSupersededAppUpdateFixture(t, true, true)
fixture := prepareSupersededAppUpdateFixture(t, true, true)
installed, err := repairPlanTreeContentStateID(fixture.target)
if err != nil {
t.Fatal(err)
}
recordHandoff(t, fixture.transaction, 1<<30, installed)
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.19.7")
if err != nil || archived {
t.Fatalf("modern transaction archived=%v err=%v", archived, err)
@@ -149,23 +171,66 @@ func TestArchiveSupersededPendingAppBundleUpdateLeavesModernRollbackForHealthCom
func TestArchiveSupersededPendingAppBundleUpdateLeavesCurrentProcessHandoff(t *testing.T) {
fixture := prepareSupersededAppUpdateFixture(t, false, false)
fixture.transaction.HandoffOwnerPID = os.Getpid()
fixture.transaction.HandoffStagingPath = filepath.Join(filepath.Dir(fixture.target), "handoff-stage")
fixture.transaction.HandoffAppPath = filepath.Join(fixture.transaction.HandoffStagingPath, "Reasonix.app")
fixture.transaction.HandoffAppTreeID = strings.Repeat("b", 64)
fixture.transaction.HandoffStagingTreeID = strings.Repeat("c", 64)
if err := overwritePendingUpdateForTest(fixture.transaction); err != nil {
t.Fatal(err)
}
recordHandoff(t, fixture.transaction, os.Getpid(), strings.Repeat("b", 64))
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.19.5")
if err != nil || archived {
t.Fatalf("current handoff archived=%v err=%v", archived, err)
}
if _, err := readPendingUpdateUnchecked(); err != nil {
if _, err := ReadPendingUpdate(); err != nil {
t.Fatalf("current handoff transaction changed: %v", err)
}
}
// In host mode the recorded owner is the Electron parent, not the service
// process that runs the shutdown archival (#10732).
func TestArchiveSupersededPendingAppBundleUpdateLeavesHostParentHandoff(t *testing.T) {
fixture := prepareSupersededAppUpdateFixture(t, false, false)
recordHandoff(t, fixture.transaction, os.Getppid(), strings.Repeat("b", 64))
eligible, err := validateSupersededPendingAppBundleUpdate(fixture.transaction, "v1.19.5")
if err != nil || eligible {
t.Fatalf("host-parent handoff eligible=%v err=%v", eligible, err)
}
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.19.5")
if err != nil || archived {
t.Fatalf("host-parent handoff archived=%v err=%v", archived, err)
}
if _, err := ReadPendingUpdate(); err != nil {
t.Fatalf("host-parent handoff transaction changed: %v", err)
}
}
func TestArchiveSupersededPendingAppBundleUpdateRetiresUnverifiableBackupAtTarget(t *testing.T) {
fixture := prepareSupersededAppUpdateFixture(t, true, true)
if err := os.WriteFile(filepath.Join(fixture.backup, ".DS_Store"), []byte("finder"), 0o600); err != nil {
t.Fatal(err)
}
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.19.7")
if err != nil || !archived {
t.Fatalf("archived=%v err=%v", archived, err)
}
if _, err := os.Lstat(PendingUpdatePath()); !os.IsNotExist(err) {
t.Fatalf("pending transaction survived: %v", err)
}
backups, err := filepath.Glob(fixture.backup + ".reasonix-retired-*")
if err != nil || len(backups) != 1 {
t.Fatalf("archived backups=%v err=%v", backups, err)
}
if _, err := os.Stat(filepath.Join(backups[0], ".DS_Store")); err != nil {
t.Fatalf("archived backup lost its contents: %v", err)
}
}
func TestArchiveSupersededPendingAppBundleUpdateRetiresSurvivingBackupPastTarget(t *testing.T) {
prepareSupersededAppUpdateFixture(t, true, true)
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.20.0")
if err != nil || !archived {
t.Fatalf("archived=%v err=%v", archived, err)
}
if _, err := os.Lstat(PendingUpdatePath()); !os.IsNotExist(err) {
t.Fatalf("pending transaction survived: %v", err)
}
}
func TestArchiveSupersededPendingAppBundleUpdateRejectsUnrelatedOlderVersion(t *testing.T) {
fixture := prepareSupersededAppUpdateFixture(t, true, false)
archived, err := ArchiveSupersededPendingAppBundleUpdate("v1.19.4")
+3 -1
View File
@@ -733,6 +733,7 @@ const __T = {
'use_model': 'Use',
'no_sessions': 'No sessions',
'error_loading': 'Error loading',
'resume_failed': 'Could not switch to that session',
'no_checkpoints': 'No checkpoints available.',
'no_conversation': 'Start a conversation before using this action.',
'no_branches': 'No conversation branches yet.',
@@ -885,6 +886,7 @@ const __T = {
'use_model': '使用',
'no_sessions': '暂无会话',
'error_loading': '加载失败',
'resume_failed': '无法切换到该会话',
'no_checkpoints': '暂无可用检查点',
'no_conversation': '先开始一段对话,再使用此操作。',
'no_branches': '暂无会话分支',
@@ -1742,7 +1744,7 @@ function renderSessions(){
const title=sessionTitle(s);
const meta=s.turns?s.turns+' turns':'';
item.innerHTML='<svg class="session-item__icon" width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><path d="M21 15a2 2 0 0 1-2 2H7l-4 4V5a2 2 0 0 1 2-2h14a2 2 0 0 1 2 2z"/></svg><div class="session-item__body"><div class="session-item__title">'+escHtml(title)+'</div><div class="session-item__meta">'+escHtml(meta)+'</div></div><button type="button" class="session-del" data-name="'+escAttr(s.name)+'" title="'+escAttr(__('delete_session'))+'">&times;</button>';
item.onclick=ev=>{if(ev.target.closest('.session-del')||running||s.current)return;post('/resume',{path:s.path}).then(()=>{log.innerHTML='';log.appendChild(welcome);showWelcome();hasVisibleHistory=false;checkpointCount=0;todosDismissed=false;resetCumulativeStats();loadSessions();updateActionAvailability();setTimeout(()=>fetch('/history').then(r=>r.json()).then(msgs=>{renderHistoryMessages(msgs);fetchTodos();refreshCheckpointAvailability();}),300);});};
item.onclick=ev=>{if(ev.target.closest('.session-del')||running||s.current)return;post('/resume',{path:s.path,sessionId:s.sessionId,hostId:s.hostId,name:s.name}).then(async r=>{if(!r.ok){const detail=(await r.text().catch(()=>'')).trim();showNotice(__('resume_failed')+' (HTTP '+r.status+(detail?': '+detail:'')+')','warn');return;}log.innerHTML='';log.appendChild(welcome);showWelcome();hasVisibleHistory=false;checkpointCount=0;todosDismissed=false;resetCumulativeStats();loadSessions();updateActionAvailability();setTimeout(()=>fetch('/history').then(r=>r.json()).then(msgs=>{renderHistoryMessages(msgs);fetchTodos();refreshCheckpointAvailability();}),300);}).catch(()=>showNotice(__('resume_failed'),'warn'));};
list.appendChild(item);
});
}
+22
View File
@@ -0,0 +1,22 @@
package serve
import (
"strings"
"testing"
)
// Canonical /sessions rows carry an empty path, so a resume that sends only
// the path cannot name the row the user clicked.
func TestServeIndexResumesSessionByIdentity(t *testing.T) {
html := string(indexHTML)
for _, want := range []string{
"post('/resume',{path:s.path,sessionId:s.sessionId,hostId:s.hostId,name:s.name})",
"if(!r.ok){const detail=(await r.text().catch(()=>'')).trim();showNotice(__('resume_failed')",
"'resume_failed': 'Could not switch to that session'",
"'resume_failed': '无法切换到该会话'",
} {
if !strings.Contains(html, want) {
t.Fatalf("serve index missing session resume handling %q", want)
}
}
}
@@ -0,0 +1,142 @@
package session
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"testing"
"reasonix/internal/provider"
)
func duplicateReplacementPayload(t *testing.T, kind string) json.RawMessage {
t.Helper()
messages := []provider.Message{
{ID: "dup", Role: provider.RoleUser, Content: "first dup"},
{ID: "answer", Role: provider.RoleAssistant, Content: "plan"},
{ID: "dup", Role: provider.RoleUser, Content: "second dup"},
{ID: "tail", Role: provider.RoleAssistant, Content: "done"},
}
body := map[string]any{"messages": messages, "reason": "fresh-resume"}
if kind == "legacy/import" {
body = map[string]any{"messages": messages, "source": Source{Path: "legacy.jsonl"}}
}
payload, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
return payload
}
func TestReplacementRefusesDuplicateMessageID(t *testing.T) {
for _, kind := range []string{"history/replace", "legacy/import"} {
t.Run(kind, func(t *testing.T) {
service, err := NewService("local", NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions")))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "dup-write"})
if err != nil {
t.Fatal(err)
}
_, err = runtime.Session().Append(t.Context(), Batch{OperationID: "dup", Events: []Event{{Kind: kind, Payload: duplicateReplacementPayload(t, kind)}}})
if !errors.Is(err, ErrDuplicateMessageID) {
t.Fatalf("append duplicate %s: %v, want ErrDuplicateMessageID", kind, err)
}
})
}
}
// commitDuplicateReplacement writes a history/replace the writer now refuses,
// standing in for a log committed before that check existed.
func commitDuplicateReplacement(t *testing.T, s *Session) {
t.Helper()
valid, err := json.Marshal(map[string]any{"messages": []provider.Message{{ID: "placeholder", Role: provider.RoleUser, Content: "x"}}})
if err != nil {
t.Fatal(err)
}
prepared, err := s.PrepareBatchContext(t.Context(), "dup-on-disk", Batch{Events: []Event{{Kind: "history/replace", Payload: valid}}})
if err != nil {
t.Fatal(err)
}
payload := duplicateReplacementPayload(t, "history/replace")
prepared.events[0].Payload = payload
prepared.storedEvents[0].Payload, prepared.storedEvents[0].PayloadRef = payload, nil
if _, err := s.CommitPrepared(prepared); err != nil {
t.Fatal(err)
}
if _, err := s.Flush(t.Context()); err != nil {
t.Fatal(err)
}
}
func TestDuplicateMessageIDOnDiskStillHydrates(t *testing.T) {
root := filepath.Join(t.TempDir(), "sessions")
service, err := NewService("local", NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "dup-read"})
if err != nil {
t.Fatal(err)
}
commitDuplicateReplacement(t, runtime.Session())
want := []string{"dup", "answer", "tail"}
if got := messageIDs(runtime.Session().Snapshot().Projection.Messages); !idsEqual(got, want) {
t.Errorf("projection ids = %v, want %v", got, want)
}
ref := runtime.Ref()
if err := service.CloseAll(t.Context()); err != nil {
t.Fatal(err)
}
for _, removeCache := range []bool{false, true} {
if removeCache {
if err := os.RemoveAll(filepath.Join(root, ".query-cache")); err != nil {
t.Fatal(err)
}
}
reopened, err := NewService("local", NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
query := reopened.Query()
if _, _, err := query.prepareHistoryIndex(t.Context(), ref); err != nil {
t.Fatalf("prepare history index (cache removed=%v): %v", removeCache, err)
}
page, err := query.ReadHistoryWindow(t.Context(), ref, HistoryWindowRequest{Anchor: "newest"})
if err != nil || page.Status != "ready" {
t.Fatalf("read history (cache removed=%v): %+v, %v", removeCache, page, err)
}
if got := windowIDs(t, page); !idsEqual(got, want) {
t.Fatalf("history ids (cache removed=%v) = %v, want %v", removeCache, got, want)
}
if search := searchHistoryReady(t, query, ref, "dup", "", 10); len(search.Hits) != 1 {
t.Fatalf("search (cache removed=%v): %+v", removeCache, search)
}
opened, err := reopened.Open(t.Context(), ref)
if err != nil {
t.Fatalf("reopen (cache removed=%v): %v", removeCache, err)
}
if got := messageIDs(opened.Runtime().Session().Snapshot().Projection.Messages); !idsEqual(got, want) {
t.Fatalf("reopened projection ids = %v, want %v", got, want)
}
if err := opened.Release(t.Context()); err != nil {
t.Fatal(err)
}
if err := reopened.CloseAll(t.Context()); err != nil {
t.Fatal(err)
}
}
}
func messageIDs(messages []provider.Message) []string {
ids := make([]string, 0, len(messages))
for _, message := range messages {
ids = append(ids, message.ID)
}
return ids
}
+61
View File
@@ -2,6 +2,7 @@ package session
import (
"encoding/json"
"errors"
"fmt"
"strings"
@@ -47,7 +48,21 @@ type legacyImportPayload struct {
ModelIdentity string `json:"modelIdentity,omitempty"`
}
// ErrDuplicateMessageID refuses a write that would give two messages one
// stable id; every reader of the log keys a message by that id.
var ErrDuplicateMessageID = errors.New("session: duplicate stable message id")
// replacementEventMessages keeps the first occurrence of each id, so a log
// written before the writer refused duplicates projects and indexes alike.
func replacementEventMessages(event Event, payload json.RawMessage) ([]provider.Message, error) {
messages, err := decodeReplacementMessages(event, payload)
if err != nil {
return nil, err
}
return firstMessageOccurrences(messages), nil
}
func decodeReplacementMessages(event Event, payload json.RawMessage) ([]provider.Message, error) {
var messages []provider.Message
var err error
if event.Kind == "legacy/import" {
@@ -64,3 +79,49 @@ func replacementEventMessages(event Event, payload json.RawMessage) ([]provider.
}
return messages, nil
}
func firstMessageOccurrences(messages []provider.Message) []provider.Message {
if _, duplicated := duplicateMessageID(messages); !duplicated {
return messages
}
seen := make(map[string]bool, len(messages))
unique := make([]provider.Message, 0, len(messages))
for _, message := range messages {
if message.ID != "" && seen[message.ID] {
continue
}
seen[message.ID] = true
unique = append(unique, message)
}
return unique
}
func duplicateMessageID(messages []provider.Message) (string, bool) {
seen := make(map[string]bool, len(messages))
for _, message := range messages {
if message.ID == "" {
continue
}
if seen[message.ID] {
return message.ID, true
}
seen[message.ID] = true
}
return "", false
}
// checkReplacementIdentities refuses a new replacement that repeats an id. A
// payload that does not decode is the projection's to refuse.
func checkReplacementIdentities(event Event) error {
if event.Kind != "history/replace" && event.Kind != "legacy/import" {
return nil
}
messages, err := decodeReplacementMessages(event, event.Payload)
if err != nil {
return nil
}
if id, duplicated := duplicateMessageID(messages); duplicated {
return fmt.Errorf("%w: %s message %q", ErrDuplicateMessageID, event.Kind, id)
}
return nil
}
+7 -5
View File
@@ -200,8 +200,9 @@ func projectLegacyImport(projection *Projection, commit Commit, ev Event) error
if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil {
return damagedPayload(ev, err)
}
projection.Messages = append([]provider.Message{}, body.Messages...)
projection.ModelMessages = append([]provider.Message{}, provider.ModelMessages(body.Messages)...)
messages := firstMessageOccurrences(body.Messages)
projection.Messages = append([]provider.Message{}, messages...)
projection.ModelMessages = append([]provider.Message{}, provider.ModelMessages(messages)...)
projection.GoalState = cloneRaw(body.Goal)
projection.ModelRef = strings.TrimSpace(body.ModelRef)
projection.ModelIdentity = strings.TrimSpace(body.ModelIdentity)
@@ -216,7 +217,7 @@ func projectMessageComplete(projection *Projection, commit Commit, ev Event) err
return damagedPayload(ev, err)
}
if projectionMessageIndex(projection.Messages, body.Message.ID) >= 0 {
return damagedPayload(ev, fmt.Errorf("duplicate stable message id %q", body.Message.ID))
return damagedPayload(ev, fmt.Errorf("%w %q", ErrDuplicateMessageID, body.Message.ID))
}
projection.Messages = append(projection.Messages, *body.Message)
projection.ModelMessages = append(projection.ModelMessages, provider.ModelMessages([]provider.Message{*body.Message})...)
@@ -314,8 +315,9 @@ func projectHistoryReplace(projection *Projection, commit Commit, ev Event) erro
if err := strictPayload(ev.Payload, &body); err != nil || body.Messages == nil {
return damagedPayload(ev, err)
}
projection.Messages = append([]provider.Message(nil), body.Messages...)
projection.ModelMessages = append([]provider.Message(nil), provider.ModelMessages(body.Messages)...)
messages := firstMessageOccurrences(body.Messages)
projection.Messages = append([]provider.Message(nil), messages...)
projection.ModelMessages = append([]provider.Message(nil), provider.ModelMessages(messages)...)
return nil
}
+3
View File
@@ -170,6 +170,9 @@ func (s *Session) PrepareBatchContext(ctx context.Context, operationID string, b
if !event.Optional && !ProjectionKinds[event.Kind] {
return PreparedBatch{}, fmt.Errorf("%w: unknown required event %q", ErrUnsupportedVersion, event.Kind)
}
if err := checkReplacementIdentities(*event); err != nil {
return PreparedBatch{}, err
}
// The durable sequence is assigned at commit time so a rejected batch
// never consumes one.
event.Sequence = 0
+3
View File
@@ -2,6 +2,8 @@ package sessioncatalog
import (
"context"
"reasonix/internal/projectiondb"
)
func (c *Catalog) Close(ctx context.Context) error {
@@ -16,6 +18,7 @@ func (c *Catalog) Close(ctx context.Context) error {
go func() {
c.closeReadLeases()
c.workers.Wait()
projectiondb.CheckpointBeforeClose(context.Background(), c.db)
c.closeErr = c.db.Close()
c.statusMu.Lock()
c.status.State = StateClosed
+27 -5
View File
@@ -49,18 +49,40 @@ func projectSessionHeads(info agent.SessionOrderInfo) sessionHeadProjection {
}
out.selected = idx.SelectedHead
out.headCount = len(idx.Heads)
out.heads = make([]HeadRecord, 0, len(idx.Heads))
out.heads, out.fingerprint = headRecordsFromIndex(info.Path, idx)
return out
}
func headRecordsFromIndex(path string, idx *agent.SessionHeadIndex) ([]HeadRecord, string) {
heads := make([]HeadRecord, 0, len(idx.Heads))
fingerprint := ""
for _, h := range idx.Heads {
out.heads = append(out.heads, HeadRecord{
Path: info.Path, ID: h.ID, ParentHeadID: h.ParentHead, Kind: h.Kind, Name: h.Name,
heads = append(heads, HeadRecord{
Path: path, ID: h.ID, ParentHeadID: h.ParentHead, Kind: h.Kind, Name: h.Name,
LeafMessageID: h.LeafID, WriterID: h.Writer, LastActivityAt: unixMilli(h.LastActivity),
Turns: h.Turns, Preview: h.Preview, Retired: h.Retired, Selected: h.ID == idx.SelectedHead,
})
if h.ID == idx.SelectedHead {
out.fingerprint = "|h:" + h.ID + ":" + h.LeafID
fingerprint = "|h:" + h.ID + ":" + h.LeafID
}
}
return out
return heads, fingerprint
}
// projectRepairedHeads lands the head rows of a session whose repair just
// completed inside that transaction: repair rebuilds a stale head index, and
// the row must not turn valid while its head_count names rows it lacks.
func projectRepairedHeads(ctx context.Context, tx *sql.Tx, path, pathKey string) error {
idx, err := agent.ReadSessionHeadIndex(path)
if err != nil || idx == nil || !idx.Current(path) {
return nil
}
heads, _ := headRecordsFromIndex(path, idx)
if _, err := tx.ExecContext(ctx, `UPDATE catalog_sessions SET log_format=?,head_count=?,selected_head_id=?
WHERE path_key=?`, idx.SchemaVersion, len(heads), idx.SelectedHead, pathKey); err != nil {
return err
}
return upsertHeadRows(ctx, tx, pathKey, heads)
}
// writeDirectoryRow lands one scanned session row and its head rows inside
@@ -128,6 +128,50 @@ func TestReconcileMarksStaleHeadIndexUnknownAndKeepsRows(t *testing.T) {
}
}
func TestFreshCatalogOverStaleHeadIndexProjectsEveryHead(t *testing.T) {
ctx := context.Background()
dir := t.TempDir()
path := filepath.Join(dir, "heads.jsonl")
writeSchemaTwoSession(t, path)
// An older build appends to the log without rewriting the head index.
if err := os.Remove(store.SessionEventIndex(path)); err != nil {
t.Fatal(err)
}
catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
target := DirectoryTarget{Path: dir, Scope: "global"}
assertHeadRowsMatchCount := func(stage string) {
t.Helper()
var rows, declared int
if err := catalog.db.QueryRowContext(ctx, `SELECT (SELECT COUNT(*) FROM catalog_heads),
(SELECT COALESCE(SUM(head_count),0) FROM catalog_sessions)`).Scan(&rows, &declared); err != nil {
t.Fatal(err)
}
if rows != declared || rows != 2 {
t.Fatalf("%s: catalog_heads rows=%d, sum(head_count)=%d, want 2 and 2", stage, rows, declared)
}
page, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "global", Limit: 10})
if err != nil || len(page.Items) != 1 || page.Items[0].Path != path {
t.Fatalf("%s: sessions = %+v err=%v", stage, page.Items, err)
}
}
if err := catalog.ReconcileDirectory(ctx, target); err != nil {
t.Fatal(err)
}
catalog.runRepairWave(ctx)
assertHeadRowsMatchCount("after repair")
if rec, ok, err := catalog.GetSession(ctx, path); err != nil || !ok || rec.TurnsState != TurnsValid {
t.Fatalf("repaired session = %+v ok=%v err=%v, want valid", rec, ok, err)
}
if err := catalog.ReconcileDirectory(ctx, target); err != nil {
t.Fatal(err)
}
assertHeadRowsMatchCount("after reconcile")
}
func TestMigrationV12AddsHeadProjectionAndForcesRescan(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "catalog.sqlite")
+3
View File
@@ -391,6 +391,9 @@ func (c *Catalog) applyGuardedRepairBatch(ctx context.Context, outcomes []repair
}
mutated++
if state == "complete" {
if err := projectRepairedHeads(ctx, tx, outcome.item.path, outcome.item.pathKey); err != nil {
return rollback(err)
}
committedDirty[queuePathKey(outcome.item.target.Path)] = outcome.item.target
}
if outcome.item.topicID != "" {
@@ -0,0 +1,28 @@
package builtin
import (
"encoding/json"
"testing"
"reasonix/internal/tool"
)
func TestBuiltinContractSchemasDeclareRequiredArray(t *testing.T) {
seen := false
for _, entry := range tool.BuiltinContractEntries() {
var root map[string]json.RawMessage
if err := json.Unmarshal(entry.Schema, &root); err != nil {
t.Fatalf("%s: schema is not an object: %v", entry.Name, err)
}
var required []string
if raw, ok := root["required"]; !ok || string(raw) == "null" || json.Unmarshal(raw, &required) != nil {
t.Errorf("%s: root schema lacks a required array: %s", entry.Name, entry.Schema)
}
if entry.Name == "get_goal" {
seen = true
}
}
if !seen {
t.Fatal("get_goal is not a registered builtin")
}
}