Compare commits

...
Author SHA1 Message Date
“haonan” 38b0debba9 fix(test): share one BSK_HOME lock across lib tests 2026-08-24 10:40:37 +08:00
“haonan” 4677149e5e fix(record): recover failed Trace exports and hint --url when the default start page fails
Reject legacy JSON --output before recording starts, persist the completed Trace before writing the bundle so `bsk record stop --output <dir>` can retry after an export failure, and tell users to pass --url when https://example.com/ does not load.
2026-08-23 21:26:40 +08:00
Ljy-0827 8ea5477e11 fix(record): safety problem and bug fix 2026-08-23 21:26:40 +08:00
“haonan” 17af1af65a feat(record): export Trace v3 bundles with settled page states 2026-08-23 21:26:40 +08:00
46 changed files with 4634 additions and 300 deletions
@@ -1,10 +1,12 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { RECORD_STOP, type RecordStepPayload } from "@/lib/record-bridge";
import { RECORD_START, RECORD_STOP, type RecordStepPayload } from "@/lib/record-bridge";
import { handleRecordContentMessage, startRecordCapture } from "../record-capture";
vi.stubGlobal("chrome", {
runtime: {
sendMessage: vi.fn(() => Promise.resolve()),
sendMessage: vi.fn((message: { sequence?: number }) =>
Promise.resolve({ ok: true, sequence: message.sequence }),
),
},
});
@@ -96,6 +98,66 @@ describe("handleRecordContentMessage stop/cancel", () => {
expect(dispose).not.toHaveBeenCalled();
expect(onStop).not.toHaveBeenCalled();
});
it("keeps a failed STOP retryable and redelivers its recorded step", async () => {
document.body.innerHTML = `<label for="retry">Draft</label><input id="retry" />`;
const sendMessage = vi.mocked(chrome.runtime.sendMessage);
sendMessage.mockReset();
sendMessage.mockRejectedValueOnce(new Error("service worker unavailable"));
sendMessage.mockRejectedValueOnce(new Error("service worker still unavailable"));
sendMessage.mockResolvedValueOnce({ ok: true, sequence: 1 });
let activeRequestId: string | null = null;
let capture: ReturnType<typeof startRecordCapture> | null = null;
const onStop = vi.fn();
const dispatch = (
message:
| { type: typeof RECORD_START; requestId: string; startedAtMs?: number }
| { type: typeof RECORD_STOP; requestId: string },
sendResponse?: (response: unknown) => void,
) =>
handleRecordContentMessage(
message,
{
activeRequestId,
capture,
setActiveRequestId: (id) => {
activeRequestId = id;
},
setCapture: (next) => {
capture = next;
},
onStart: vi.fn(),
onStop,
},
sendResponse,
);
dispatch({ type: RECORD_START, requestId: "rec-retry" });
const input = document.querySelector("input")!;
input.dispatchEvent(new FocusEvent("focusin", { bubbles: true }));
input.value = "dirty final value";
input.dispatchEvent(new Event("input", { bubbles: true }));
const firstResponse = vi.fn();
expect(dispatch({ type: RECORD_STOP, requestId: "rec-retry" }, firstResponse)).toBe(true);
await vi.waitFor(() =>
expect(firstResponse).toHaveBeenCalledWith({
ok: false,
error: "failed to deliver one or more recorded steps",
}),
);
expect(activeRequestId).toBe("rec-retry");
expect(onStop).not.toHaveBeenCalled();
const retryResponse = vi.fn();
expect(dispatch({ type: RECORD_STOP, requestId: "rec-retry" }, retryResponse)).toBe(true);
await vi.waitFor(() => expect(retryResponse).toHaveBeenCalledWith({ ok: true }));
expect(sendMessage).toHaveBeenCalledTimes(3);
expect(activeRequestId).toBeNull();
expect(onStop).toHaveBeenCalledTimes(1);
});
});
describe("record-capture semantic", () => {
@@ -228,6 +290,7 @@ describe("record-capture semantic", () => {
expect(steps[0]).toMatchObject({
op: "hover",
target: { role: "button", name: "Open user navigation menu" },
geometry: { tag: "button", rect: { x: 900, y: 8, w: 32, h: 32 } },
});
expect(steps[1]).toMatchObject({
op: "click",
@@ -0,0 +1,36 @@
import { describe, expect, it, vi } from "vitest";
import type { RecordStepMessage } from "@/lib/record-bridge";
import { RecordStepDelivery } from "../record-step-delivery";
describe("RecordStepDelivery", () => {
it("retries from the first unacknowledged sequence before sending later steps", async () => {
const sent: RecordStepMessage[] = [];
const send = vi.fn(async (message: RecordStepMessage) => {
sent.push(message);
if (sent.length === 1) throw new Error("service worker unavailable");
return { ok: true, sequence: message.sequence };
});
const delivery = new RecordStepDelivery("rec-ordered", send, "document-1");
delivery.enqueue({ op: "click", target: { tag: "button", name: "First" } });
delivery.enqueue({ op: "click", target: { tag: "button", name: "Second" } });
await expect(delivery.flush()).resolves.toBe(true);
expect(sent.map((message) => message.sequence)).toEqual([1, 1, 2]);
expect(sent.map((message) => message.step.target?.name)).toEqual(["First", "First", "Second"]);
});
it("keeps unacknowledged steps pending for a later flush", async () => {
const send = vi
.fn<(message: RecordStepMessage) => Promise<unknown>>()
.mockRejectedValueOnce(new Error("offline"))
.mockRejectedValueOnce(new Error("still offline"))
.mockImplementation(async (message) => ({ ok: true, sequence: message.sequence }));
const delivery = new RecordStepDelivery("rec-retry", send, "document-1");
delivery.enqueue({ op: "click", target: { tag: "button", name: "Save" } });
await expect(delivery.flush()).resolves.toBe(false);
await expect(delivery.flush()).resolves.toBe(true);
expect(send).toHaveBeenCalledTimes(3);
});
});
+68 -44
View File
@@ -36,27 +36,17 @@ import {
isHoverSurfaceCandidateElement,
isLikelyHoverSurfaceOwner,
} from "./record-hover-surface";
import { RecordStepDelivery } from "./record-step-delivery";
const pendingStepSends = new Map<string, Set<Promise<boolean>>>();
const failedStepDeliveries = new Set<string>();
const knownRecordRequests = new Set<string>();
const stepDeliveries = new Map<string, RecordStepDelivery>();
const pendingStopFlushes = new Map<string, Promise<RecordStopAck>>();
function sendRecordStep(requestId: string, step: RecordStepPayload): void {
const payload = { type: RECORD_STEP, requestId, step };
const pending = Promise.resolve(chrome.runtime.sendMessage(payload)).then(
() => true,
() => {
failedStepDeliveries.add(requestId);
return false;
},
);
const sends = pendingStepSends.get(requestId) ?? new Set<Promise<boolean>>();
sends.add(pending);
pendingStepSends.set(requestId, sends);
void pending.then(() => {
sends.delete(pending);
if (sends.size === 0) pendingStepSends.delete(requestId);
});
function deliveryFor(requestId: string): RecordStepDelivery {
const existing = stepDeliveries.get(requestId);
if (existing) return existing;
const created = new RecordStepDelivery(requestId);
stepDeliveries.set(requestId, created);
return created;
}
export interface RecordCaptureController {
@@ -68,6 +58,7 @@ interface FillSession {
target: CaptureTargetDescriptor;
baselineValue: string;
lastValue: string;
pendingCommit?: "enter" | "suggestion" | "blur";
}
interface HoverCandidate {
@@ -157,6 +148,24 @@ function nearbyFillableFromSearchChrome(target: Element): FillableElement | null
return null;
}
function captureGeometry(el: Element): RecordStepPayload["geometry"] {
const rect = el.getBoundingClientRect();
return {
rect: { x: rect.x, y: rect.y, w: rect.width, h: rect.height },
tag: el.tagName.toLowerCase(),
};
}
function geometryForEventTarget(
target: EventTarget | null,
): RecordStepPayload["geometry"] | undefined {
if (!(target instanceof Element)) return undefined;
const clickable = target.closest(
'a,button,input,select,textarea,[role="button"],[role="link"],[role="menuitem"],[contenteditable="true"]',
);
return captureGeometry(clickable ?? target);
}
function fillableValue(el: FillableElement): string {
if (el instanceof HTMLInputElement || el instanceof HTMLTextAreaElement) {
return el.value;
@@ -414,14 +423,17 @@ export function startRecordCapture(
emitStep({
op: "fill",
target: session.target,
geometry: captureGeometry(session.element),
value,
commit: session.pendingCommit ?? "blur",
...(isPassword ? { redacted: true } : {}),
});
};
const commitFillSession = () => {
const commitFillSession = (commit: "enter" | "suggestion" | "blur" = "blur") => {
if (!fillSession || composing) return;
const session = fillSession;
session.pendingCommit = commit;
fillSession = null;
emitFill(session);
committedValues.set(session.element, session.lastValue);
@@ -627,6 +639,7 @@ export function startRecordCapture(
emitStep({
op: "hover",
target: hover.target,
geometry: captureGeometry(hover.element),
});
emittedHoverElements.add(hover.element);
};
@@ -681,6 +694,7 @@ export function startRecordCapture(
emitStep({
op: "click",
target,
geometry: geometryForEventTarget(eventTarget(event)),
expects_navigation: true,
});
};
@@ -782,7 +796,7 @@ export function startRecordCapture(
scheduleInputCompletionCommit(
sessionElement,
syncFillSessionValue,
commitFillSession,
() => commitFillSession("suggestion"),
(el) => fillSession?.element === el,
);
return;
@@ -842,6 +856,7 @@ export function startRecordCapture(
emitStep({
op: "select",
target: desc,
geometry: captureGeometry(target),
values,
labels,
expects_navigation: true,
@@ -882,7 +897,7 @@ export function startRecordCapture(
};
}
if (fillable) {
commitFillSession();
commitFillSession(event.key === "Enter" ? "enter" : "blur");
}
const desc = describeEventTarget(target);
if (!desc && !event.key) return;
@@ -890,7 +905,7 @@ export function startRecordCapture(
emitStep({
op: "press",
key: event.key,
...(desc ? { target: desc } : {}),
...(desc ? { target: desc, geometry: geometryForEventTarget(target) } : {}),
...(modifiers.length ? { modifiers } : {}),
expects_navigation: event.key === "Enter",
});
@@ -909,9 +924,12 @@ export function startRecordCapture(
const urlObserver = new MutationObserver(() => emitNavigateIfChanged());
urlObserver.observe(document, { subtree: true, childList: true });
const onUrlEvent = () => emitNavigateIfChanged();
// Wrap: passing commitFillSession directly would forward the DOM event as
// the `commit` argument and stamp it onto the recorded fill step.
const onPageHide = () => commitFillSession();
window.addEventListener("hashchange", onUrlEvent);
window.addEventListener("popstate", onUrlEvent);
window.addEventListener("pagehide", commitFillSession);
window.addEventListener("pagehide", onPageHide);
const originalPushState = history.pushState;
const originalReplaceState = history.replaceState;
@@ -939,7 +957,7 @@ export function startRecordCapture(
urlObserver.disconnect();
window.removeEventListener("hashchange", onUrlEvent);
window.removeEventListener("popstate", onUrlEvent);
window.removeEventListener("pagehide", commitFillSession);
window.removeEventListener("pagehide", onPageHide);
history.pushState = originalPushState;
history.replaceState = originalReplaceState;
},
@@ -965,14 +983,11 @@ export function handleRecordContentMessage(
sendResponse?: (response: RecordStartAck | RecordStopAck) => void,
): boolean {
if (isRecordStartMessage(message)) {
if (!knownRecordRequests.has(message.requestId)) {
knownRecordRequests.add(message.requestId);
failedStepDeliveries.delete(message.requestId);
}
const delivery = deliveryFor(message.requestId);
state.capture?.dispose();
state.setCapture(
startRecordCapture(message.requestId, (step) => {
sendRecordStep(message.requestId, step);
delivery.enqueue(step);
}),
);
state.setActiveRequestId(message.requestId);
@@ -995,28 +1010,37 @@ export function handleRecordContentMessage(
state.setActiveRequestId(null);
};
if (isRecordStopMessage(message) && sendResponse) {
const pending = [...(pendingStepSends.get(message.requestId) ?? [])];
void Promise.all(pending).then((delivered) => {
finishStop();
const succeeded = delivered.every(Boolean) && !failedStepDeliveries.has(message.requestId);
if (succeeded) {
failedStepDeliveries.delete(message.requestId);
knownRecordRequests.delete(message.requestId);
}
sendResponse(
succeeded
const existingFlush = pendingStopFlushes.get(message.requestId);
if (existingFlush) {
void existingFlush.then(sendResponse);
return true;
}
const flush = deliveryFor(message.requestId)
.flush()
.then((succeeded): RecordStopAck => {
if (succeeded) {
stepDeliveries.delete(message.requestId);
finishStop();
}
return succeeded
? { ok: true }
: {
ok: false,
error: "failed to deliver one or more recorded steps",
},
);
};
});
pendingStopFlushes.set(message.requestId, flush);
void flush.then(sendResponse).finally(() => {
if (pendingStopFlushes.get(message.requestId) === flush) {
pendingStopFlushes.delete(message.requestId);
}
});
return true;
}
if (isRecordCancelMessage(message)) {
failedStepDeliveries.delete(message.requestId);
knownRecordRequests.delete(message.requestId);
stepDeliveries.delete(message.requestId);
pendingStopFlushes.delete(message.requestId);
}
finishStop();
return false;
@@ -0,0 +1,65 @@
import {
isAcceptedRecordStepAck,
RECORD_STEP,
type RecordStepMessage,
type RecordStepPayload,
} from "@/lib/record-bridge";
export type SendRecordStepMessage = (message: RecordStepMessage) => Promise<unknown>;
export class RecordStepDelivery {
readonly #requestId: string;
readonly #producerId: string;
readonly #send: SendRecordStepMessage;
readonly #pending = new Map<number, RecordStepPayload>();
#nextSequence = 1;
#tail = Promise.resolve();
constructor(
requestId: string,
send: SendRecordStepMessage = (message) => chrome.runtime.sendMessage(message),
producerId: string = crypto.randomUUID(),
) {
this.#requestId = requestId;
this.#producerId = producerId;
this.#send = send;
}
enqueue(step: RecordStepPayload): void {
const sequence = this.#nextSequence;
this.#nextSequence += 1;
this.#pending.set(sequence, step);
void this.#schedule();
}
async flush(): Promise<boolean> {
await this.#tail;
await this.#schedule();
return this.#pending.size === 0;
}
#schedule(): Promise<void> {
const drain = () => this.#drain();
this.#tail = this.#tail.then(drain, drain);
return this.#tail;
}
async #drain(): Promise<void> {
for (const [sequence, step] of this.#pending) {
const message: RecordStepMessage = {
type: RECORD_STEP,
requestId: this.#requestId,
producerId: this.#producerId,
sequence,
step,
};
try {
const response = await this.#send(message);
if (!isAcceptedRecordStepAck(response, sequence)) return;
this.#pending.delete(sequence);
} catch {
return;
}
}
}
}
+2 -1
View File
@@ -172,6 +172,7 @@ export default defineBackground(() => {
dispatcher.start();
const recordDeps = {
tabsApi: chrome.tabs,
cdp,
sendToTab: (tabId: number, msg: Parameters<typeof chrome.tabs.sendMessage>[1]) =>
chrome.tabs.sendMessage(tabId, msg),
bypassOverlay: async (tabId: number, enabled: boolean) => {
@@ -188,7 +189,7 @@ export default defineBackground(() => {
// Message listeners stay up (cheap; fire only for record message types).
// Tab / webNavigation observation attaches lazily while a recording is
// active — see ensureBrowserObservationListeners in tools/record.ts.
attachRecordStepListener();
attachRecordStepListener(recordDeps);
attachRecordFinishListener(recordDeps);
attachRecordQueryListener(recordDeps);
if (typeof chrome.notifications?.onClicked?.addListener === "function") {
@@ -200,6 +200,9 @@ describe("App", () => {
expect(copied).toContain("--browser 03c3e47f");
expect(copied).toContain('--purpose "发布 wiki 文档"');
expect(copied).not.toMatch(/bsk record start[^\n]*--url/);
expect(copied).toContain("./trace");
expect(copied).toContain("trace.json");
expect(copied).toContain("states/");
// Button label stays static; a transient toast confirms the copy.
expect(copyButton.textContent).toContain("复制录制指令");
await waitFor(() => expect(screen.getByRole("status").textContent).toContain("已复制"));
@@ -26,20 +26,20 @@ function handshake(
}
describe("computeConnectedState (protocol-based compat)", () => {
it("returns connected when protocol strings match", () => {
expect(computeConnectedState(handshake("1.0", "1.0"), MIN_COMPATIBLE_PROTOCOL)).toEqual({
it("returns connected when daemon protocol equals extension protocol", () => {
expect(computeConnectedState(handshake("1.1", "1.0"), MIN_COMPATIBLE_PROTOCOL)).toEqual({
kind: "connected",
});
});
it("returns version_skew when daemon protocol minor is newer", () => {
expect(computeConnectedState(handshake("1.1", "1.0"))).toEqual({
expect(computeConnectedState(handshake("1.2", "1.0"))).toEqual({
kind: "version_skew",
});
});
it("returns version_skew when daemon protocol string differs but floor is satisfied", () => {
expect(computeConnectedState(handshake("1", "1.0"))).toEqual({
expect(computeConnectedState(handshake("1.1.0", "1.0"))).toEqual({
kind: "version_skew",
});
});
@@ -53,7 +53,7 @@ describe("computeConnectedState (protocol-based compat)", () => {
});
it("rejects when extension is below daemon min_compatible_protocol", () => {
const result = computeConnectedState(handshake("1.0", "1.5"));
const result = computeConnectedState(handshake("1.1", "1.5"));
expect(result.kind).toBe("rejected");
if (result.kind === "rejected") {
expect(result.reason).toContain("min_compatible_protocol");
@@ -65,7 +65,7 @@ describe("computeConnectedState (protocol-based compat)", () => {
const result = computeConnectedState({
server: "browser-skill-daemon",
version: "0.1.0",
protocol_version: "1.0",
protocol_version: "1.1",
min_compatible_peer: "0.1.0",
});
expect(result).toEqual({ kind: "connected" });
@@ -80,7 +80,7 @@ describe("computeConnectedState (protocol-based compat)", () => {
});
it("rejects malformed daemon min_compatible_protocol with a daemon-floor reason", () => {
const result = computeConnectedState(handshake("1.0", "not-a-protocol"));
const result = computeConnectedState(handshake("1.1", "not-a-protocol"));
expect(result.kind).toBe("rejected");
if (result.kind === "rejected") {
expect(result.reason).toContain("daemon min_compatible_protocol");
@@ -242,11 +242,11 @@ describe("ConnectionController connectionEnabled", () => {
const second = transport.send.mock.calls[1]?.[0] as { id: string };
expect(second.id).not.toBe(first.id);
transport.emitMessage({ id: first.id, result: handshake("1.0", "1.0") });
transport.emitMessage({ id: first.id, result: handshake("1.1", "1.0") });
await Promise.resolve();
expect(controller.snapshot().state).not.toBe("connected");
transport.emitMessage({ id: second.id, result: handshake("1.0", "1.0") });
transport.emitMessage({ id: second.id, result: handshake("1.1", "1.0") });
await vi.waitFor(() => expect(controller.snapshot().state).toBe("connected"));
});
});
@@ -0,0 +1,32 @@
import { describe, expect, it } from "vitest";
import {
isAcceptedRecordStepAck,
isRecordStepMessage,
RECORD_STEP,
type RecordStepMessage,
} from "../record-bridge";
const message: RecordStepMessage = {
type: RECORD_STEP,
requestId: "rec-1",
producerId: "document-1",
sequence: 1,
step: { op: "click", target: { tag: "button", name: "Submit" } },
};
describe("record step delivery protocol", () => {
it("requires a producer and a positive integer sequence", () => {
expect(isRecordStepMessage(message)).toBe(true);
expect(isRecordStepMessage({ ...message, producerId: "" })).toBe(false);
expect(isRecordStepMessage({ ...message, sequence: 0 })).toBe(false);
expect(isRecordStepMessage({ ...message, sequence: 1.5 })).toBe(false);
});
it("accepts only the acknowledgement for the sent sequence", () => {
expect(isAcceptedRecordStepAck({ ok: true, sequence: 2 }, 2)).toBe(true);
expect(isAcceptedRecordStepAck({ ok: true, sequence: 1 }, 2)).toBe(false);
expect(isAcceptedRecordStepAck({ ok: false, expectedSequence: 2, error: "gap" }, 2)).toBe(
false,
);
});
});
@@ -11,7 +11,7 @@ function sessionWithInput(redactValues = false): RecordingObservationSession {
const session = new RecordingObservationSession({ redactValues });
const state = session.registry.register({
url: URL,
rawVomText: '@vom 1\ntextbox "Password" value="••••••" [ref=e1]',
vomText: '@vom 1\ntextbox "Password" value="••••••" [ref=e1]',
});
session.cursor.lastSettled = {
stateId: state.id,
@@ -34,7 +34,7 @@ function sessionWithInput(redactValues = false): RecordingObservationSession {
return session;
}
function finalizedFillBody(value: string, redactValues: boolean): string {
function finalizedFillTrace(value: string, redactValues: boolean) {
const session = sessionWithInput(redactValues);
const draft: RecordingDraftStep = {
op: "fill",
@@ -53,15 +53,20 @@ function finalizedFillBody(value: string, redactValues: boolean): string {
startedAt: "2026-08-12T00:00:00.000Z",
stoppedBy: "user_finish",
bskVersion: "test",
}).states[0]!.body;
redactValues,
});
}
function finalizedFillBody(value: string, redactValues: boolean): string {
return finalizedFillTrace(value, redactValues).states[0]!.body;
}
describe("record observation annotations", () => {
it("omits a fill literal when values are redacted", () => {
const secret = "hunter2-private";
const body = finalizedFillBody(secret, true);
expect(body).toContain("step 1: fill");
expect(body).not.toContain(secret);
const dumped = JSON.stringify(finalizedFillTrace(secret, true));
expect(dumped).toContain("step 1: fill");
expect(dumped).not.toContain(secret);
});
it("keeps ordinary fill details", () => {
@@ -73,7 +78,7 @@ describe("record observation annotations", () => {
const state = registry.register({
url: "https://example.com/a\nb",
title: "hello\nworld",
rawVomText: "@vom 1",
vomText: "@vom 1",
});
const trace = buildTraceV3({
registry,
@@ -91,18 +96,18 @@ describe("recording state ownership", () => {
it("deduplicates within one recording and isolates ids between recordings", () => {
const first = new RecordingStateRegistry();
const second = new RecordingStateRegistry();
expect(first.register({ url: URL, rawVomText: "same" }).id).toBe("s1");
expect(first.register({ url: URL, rawVomText: "same" }).id).toBe("s1");
expect(second.register({ url: URL, rawVomText: "other" }).id).toBe("s1");
expect(first.register({ url: URL, vomText: "same" }).id).toBe("s1");
expect(first.register({ url: URL, vomText: "same" }).id).toBe("s1");
expect(second.register({ url: URL, vomText: "other" }).id).toBe("s1");
});
it("enriches metadata when a deduplicated observation becomes more complete", () => {
const registry = new RecordingStateRegistry();
registry.register({ url: URL, rawVomText: "same" });
registry.register({ url: URL, vomText: "same" });
const state = registry.register({
url: URL,
title: "Login",
rawVomText: "same",
vomText: "same",
truncated: true,
});
expect(state).toMatchObject({ id: "s1", title: "Login", truncated: true });
@@ -0,0 +1,133 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import type { CdpRunner, ChromeTabsApi } from "@/tools/shared";
const captureRecordingObservation = vi.hoisted(() => vi.fn());
vi.mock("../recording/observation-capture", async (importOriginal) => {
const actual = await importOriginal<typeof import("../recording/observation-capture")>();
return { ...actual, captureRecordingObservation };
});
import { ObservationNodeIndex } from "../recording/observation-capture";
import { RecordingObservationRuntime } from "../recording/recording-runtime";
function observation() {
return {
rootFrameId: "root",
index: new ObservationNodeIndex({ rootFrameId: "root", matchNodes: [], refs: [] }),
url: "https://example.com/",
title: "Example",
vomText: '@vom 1\nRootWebArea "Example"',
truncated: false,
};
}
function runtime(): RecordingObservationRuntime {
return new RecordingObservationRuntime({
cdp: { send: vi.fn() as unknown as CdpRunner["send"] },
tabsApi: { get: vi.fn(), query: vi.fn() } as unknown as ChromeTabsApi,
});
}
describe("RecordingObservationRuntime", () => {
afterEach(() => captureRecordingObservation.mockReset());
it("shares one initial capture and includes it in flush", async () => {
let release!: (value: ReturnType<typeof observation>) => void;
captureRecordingObservation.mockReturnValueOnce(
new Promise((resolve) => {
release = resolve;
}),
);
const recording = runtime();
const first = recording.captureInitial(7);
const second = recording.captureInitial(7);
const flushed = recording.flush();
let flushCompleted = false;
void flushed.then(() => {
flushCompleted = true;
});
await Promise.resolve();
expect(captureRecordingObservation).toHaveBeenCalledTimes(1);
expect(flushCompleted).toBe(false);
release(observation());
await Promise.all([first, second, flushed]);
expect(flushCompleted).toBe(true);
});
it("cancels an in-flight initial capture", async () => {
captureRecordingObservation.mockImplementationOnce(
(input: { signal?: AbortSignal }) =>
new Promise((_, reject) => {
input.signal?.addEventListener(
"abort",
() => reject(new DOMException("observation aborted", "AbortError")),
{ once: true },
);
}),
);
const recording = runtime();
const pending = recording.captureInitial(7);
recording.cancel();
await expect(pending).rejects.toMatchObject({ name: "AbortError" });
});
it("retries a failed initial capture when the recording settles at stop", async () => {
captureRecordingObservation
.mockRejectedValueOnce(new Error("document swapped"))
.mockResolvedValueOnce(observation());
const recording = runtime();
await expect(recording.captureInitial(7)).rejects.toThrow("document swapped");
await recording.settleTrailing(7, []);
const trace = recording.buildTrace({
drafts: [],
startedAt: "2026-08-19T00:00:00.000Z",
stoppedBy: "user_finish",
bskVersion: "0.1.6",
});
expect(captureRecordingObservation).toHaveBeenCalledTimes(2);
expect(trace.states).toHaveLength(1);
});
it("keeps runtime matching data out of the serialized trace", async () => {
const captured = observation();
captured.index = new ObservationNodeIndex({
rootFrameId: "root",
matchNodes: [
{
frameId: "root",
backendNodeId: 42,
tag: "button",
rect: { x: 10, y: 20, w: 100, h: 30 },
localRect: { x: 10, y: 20, w: 100, h: 30 },
},
],
refs: [{ ref: "e1", backendNodeId: 42, role: "button", name: "Submit", line: 1 }],
});
captureRecordingObservation.mockResolvedValueOnce(captured);
const recording = runtime();
await recording.captureInitial(7);
const dumped = JSON.stringify(
recording.buildTrace({
drafts: [],
startedAt: "2026-08-19T00:00:00.000Z",
stoppedBy: "user_finish",
bskVersion: "0.1.6",
}),
);
expect(dumped).toContain("RootWebArea");
expect(dumped).not.toContain("matchNodes");
expect(dumped).not.toContain("localRect");
expect(dumped).not.toContain("backendNodeId");
expect(dumped).not.toContain("ObservationNodeIndex");
});
});
@@ -18,6 +18,39 @@ describe("recording-step-buffer", () => {
expect(buffer.pendingNavigation).toBe(true);
});
it("keeps the hovered element description as a capture fallback", () => {
const buffer = { steps: [], pendingNavigation: false };
appendRecordedPayload(
buffer,
{
op: "hover",
target: { tag: "button", role: "button", name: "新建" },
geometry: {
rect: { x: 900, y: 8, w: 60, h: 32 },
tag: "button",
},
},
{
geometry: {
rect: { x: 900, y: 8, w: 60, h: 32 },
tag: "button",
},
},
);
expect(buffer.steps).toEqual([
{
op: "hover",
captureTarget: { tag: "button", role: "button", name: "新建" },
targetHint: {
geometry: {
rect: { x: 900, y: 8, w: 60, h: 32 },
tag: "button",
},
},
},
]);
});
it("annotates navigated_to on action-caused navigation instead of wait_for_navigation", () => {
const buffer = {
steps: [
@@ -71,10 +104,54 @@ describe("recording-step-buffer", () => {
currentUrl: "https://example.com/a",
pendingNavigation: false,
};
observeRecordedNavigation(buffer, "https://example.com/b", false);
const result = observeRecordedNavigation(buffer, "https://example.com/b", false);
expect(result).toEqual({ kind: "appended", index: 0 });
expect(buffer.steps[0]).toMatchObject({
op: "navigate",
url: "https://example.com/b",
});
});
it("asks the recorder to coalesce redirect hops instead of emitting each one", () => {
const buffer = {
steps: [],
currentUrl: "https://passport.example/login",
pendingNavigation: false,
};
const hop1 = observeRecordedNavigation(
buffer,
"https://passport.example/callback",
false,
"link",
["server_redirect"],
);
expect(hop1).toEqual({
kind: "coalesce_redirect",
url: "https://passport.example/callback",
});
expect(buffer.steps).toEqual([]);
const hop2 = observeRecordedNavigation(buffer, "https://app.example/dashboard", false, "link", [
"client_redirect",
]);
expect(hop2).toEqual({
kind: "coalesce_redirect",
url: "https://app.example/dashboard",
});
expect(buffer.steps).toEqual([]);
expect(buffer.currentUrl).toBe("https://app.example/dashboard");
});
it("does not reuse redirect metadata for a later content-observed URL change", () => {
const buffer = {
steps: [],
currentUrl: "https://example.com/redirect",
pendingNavigation: false,
};
const result = observeRecordedNavigation(buffer, "https://example.com/spa");
expect(result).toEqual({ kind: "appended", index: 0 });
expect(buffer.steps[0]).toMatchObject({ op: "navigate", url: "https://example.com/spa" });
});
});
@@ -6,6 +6,33 @@ import { reduceTraceStepsV3 } from "../recording/trace-reducer-v3";
import type { RecordingDraftStep } from "../recording/types";
describe("trace reducer v3", () => {
it("removes form literals when value redaction is requested", () => {
const reduced = reduceTraceStepsV3(
[
{
op: "fill",
value: "private@example.com",
preStateId: "s1",
postStateId: "s1",
},
{
op: "select",
values: ["private-account-id"],
labels: ["Private account"],
preStateId: "s1",
postStateId: "s1",
},
],
{ redactValues: true },
);
expect(reduced.steps[0]).toMatchObject({ op: "fill", value: "***", redacted: true });
expect(reduced.steps[1]).toMatchObject({ op: "select" });
expect(reduced.steps[1]).not.toHaveProperty("selection");
expect(JSON.stringify(reduced.steps)).not.toContain("private@example.com");
expect(JSON.stringify(reduced.steps)).not.toContain("private-account-id");
});
it("collapses redirect hops while retaining draft-to-step identity", () => {
const drafts: RecordingDraftStep[] = [
{ op: "navigate", url: "https://example.com/start", preStateId: "s1", postStateId: "s2" },
@@ -36,7 +63,7 @@ describe("trace reducer v3", () => {
it("builds the wire model from protocol constants", () => {
const registry = new RecordingStateRegistry();
const state = registry.register({ url: "https://example.com", rawVomText: "@vom 1" });
const state = registry.register({ url: "https://example.com", vomText: "@vom 1" });
const trace = buildTraceV3({
registry,
drafts: [
+37 -1
View File
@@ -16,6 +16,10 @@ export interface RecordStartAck {
ok: true;
}
export type RecordStepAck =
| { ok: true; sequence: number }
| { ok: false; expectedSequence: number; error: string };
export type RecordStopAck = { ok: true } | { ok: false; error: string };
export interface RecordQueryMessage {
@@ -50,17 +54,30 @@ export interface RecordStepPayload {
labels?: string[];
url?: string;
redacted?: boolean;
commit?: "enter" | "suggestion" | "blur";
/** Page URL when the step was captured. */
page_url?: string;
/** Event-target bounds in the source frame's viewport coordinate space. */
geometry?: {
rect: { x: number; y: number; w: number; h: number };
tag: string;
};
/** Capture-only hint; never persisted unless converted to navigated_to. */
expects_navigation?: boolean;
/** Whether an observed URL change was synchronously caused by the action. */
navigation_caused_by_action?: boolean;
/** Raw webNavigation transition metadata for cause mapping. */
transitionType?: string;
transitionQualifiers?: string[];
}
export interface RecordStepMessage {
type: typeof RECORD_STEP;
requestId: string;
/** Stable for the lifetime of one content-script document. */
producerId: string;
/** Monotonic within one content-script producer and recording request. */
sequence: number;
step: RecordStepPayload;
}
@@ -112,8 +129,27 @@ export function isRecordQueryMessage(msg: unknown): msg is RecordQueryMessage {
export function isRecordStepMessage(msg: unknown): msg is RecordStepMessage {
if (typeof msg !== "object" || msg === null) return false;
const m = msg as Record<string, unknown>;
if (m.type !== RECORD_STEP || typeof m.requestId !== "string") return false;
if (
m.type !== RECORD_STEP ||
typeof m.requestId !== "string" ||
typeof m.producerId !== "string" ||
m.producerId.length === 0 ||
typeof m.sequence !== "number" ||
!Number.isSafeInteger(m.sequence) ||
m.sequence < 1
) {
return false;
}
const step = m.step;
if (typeof step !== "object" || step === null) return false;
return typeof (step as RecordStepPayload).op === "string";
}
export function isAcceptedRecordStepAck(
value: unknown,
sequence: number,
): value is Extract<RecordStepAck, { ok: true }> {
if (typeof value !== "object" || value === null) return false;
const ack = value as Partial<RecordStepAck>;
return ack.ok === true && ack.sequence === sequence;
}
@@ -0,0 +1,5 @@
const REDIRECT_QUALIFIERS = new Set(["client_redirect", "server_redirect"]);
export function hasRedirectQualifier(qualifiers?: readonly string[]): boolean {
return (qualifiers ?? []).some((qualifier) => REDIRECT_QUALIFIERS.has(qualifier));
}
@@ -74,7 +74,7 @@ export class RecordingObservationSession {
const state = this.registry.register({
url: captured.url,
title: captured.title,
rawVomText: captured.vomText,
vomText: captured.vomText,
truncated: captured.truncated,
});
const observation: RegisteredObservation = {
@@ -0,0 +1,177 @@
import type { CdpRunner, ChromeTabsApi } from "@/tools/shared";
import type { StopReason, TraceV3 } from "@/transport/types";
import type { DocumentSettleScope } from "./document-settle";
import { RecordingObservationSession } from "./observation-session";
import { inferMissingPostStates, SettleController } from "./settle-controller";
import { RecordingStateRegistry } from "./state-registry";
import { buildTraceV3 } from "./trace-builder-v3";
import type { RecordingDraftStep, StepAnnotation } from "./types";
interface TabRecordingContext {
session: RecordingObservationSession;
settle: SettleController;
initialCapture: PendingInitialCapture | null;
}
interface PendingInitialCapture {
promise: Promise<void>;
abort: AbortController;
}
export class RecordingObservationRuntime {
readonly #cdp: CdpRunner;
readonly #tabsApi: ChromeTabsApi;
readonly #registry = new RecordingStateRegistry();
readonly #annotations: StepAnnotation[] = [];
readonly #contexts = new Map<number, TabRecordingContext>();
readonly #maxTokens?: number;
readonly #redactValues: boolean;
constructor(input: {
cdp: CdpRunner;
tabsApi: ChromeTabsApi;
maxTokens?: number;
redactValues?: boolean;
}) {
this.#cdp = input.cdp;
this.#tabsApi = input.tabsApi;
this.#maxTokens = input.maxTokens;
this.#redactValues = input.redactValues ?? false;
}
#context(tabId: number): TabRecordingContext {
const existing = this.#contexts.get(tabId);
if (existing) return existing;
const session = new RecordingObservationSession({
registry: this.#registry,
annotations: this.#annotations,
maxTokens: this.#maxTokens,
redactValues: this.#redactValues,
});
const context: TabRecordingContext = {
session,
settle: new SettleController({
session,
cdp: this.#cdp,
tabsApi: this.#tabsApi,
tabId,
}),
initialCapture: null,
};
this.#contexts.set(tabId, context);
return context;
}
async captureInitial(tabId: number): Promise<void> {
const context = this.#context(tabId);
if (context.session.cursor.lastSettled) return;
if (context.initialCapture) return context.initialCapture.promise;
const abort = new AbortController();
let pending!: PendingInitialCapture;
const promise = context.session.capture(this.#cdp, this.#tabsApi, tabId, abort.signal);
pending = {
abort,
promise: promise
.then(() => {})
.finally(() => {
if (context.initialCapture === pending) context.initialCapture = null;
}),
};
context.initialCapture = pending;
return pending.promise;
}
async processDraft(
tabId: number,
drafts: RecordingDraftStep[],
draftIndex: number,
scope?: DocumentSettleScope,
): Promise<void> {
const draft = drafts[draftIndex];
if (!draft) return;
const context = this.#context(tabId);
if (!context.session.cursor.lastSettled) {
try {
await this.captureInitial(tabId);
} catch {
// Post-action settle can still provide a usable state.
}
}
context.session.bindDraft(draft, draftIndex + 1, context.settle.hasPending);
context.settle.schedule(drafts, draftIndex, scope);
}
scheduleSettle(
tabId: number,
drafts: RecordingDraftStep[],
draftIndex: number,
scope?: DocumentSettleScope,
): void {
this.#context(tabId).settle.schedule(drafts, draftIndex, scope);
}
clearRedirect(tabId: number): void {
this.#contexts.get(tabId)?.settle.clearRedirect();
}
scheduleRedirect(tabId: number, drafts: RecordingDraftStep[], url: string): void {
this.#context(tabId).settle.scheduleRedirect(drafts, url);
}
async flushRedirects(): Promise<void> {
for (const context of this.#contexts.values()) await context.settle.flushRedirects();
}
async flush(): Promise<void> {
await Promise.allSettled(
[...this.#contexts.values()].flatMap((context) =>
context.initialCapture ? [context.initialCapture.promise] : [],
),
);
await this.flushRedirects();
for (const context of this.#contexts.values()) await context.settle.flush();
}
async settleTrailing(tabId: number, drafts: RecordingDraftStep[]): Promise<void> {
const context = this.#context(tabId);
if (context.initialCapture) await Promise.allSettled([context.initialCapture.promise]);
if (!context.session.cursor.lastSettled) {
try {
await this.captureInitial(tabId);
} catch {
// The trace may remain observation-free when CDP is unavailable.
}
}
await context.settle.settleTrailing(drafts);
inferMissingPostStates(drafts);
}
cancel(): void {
for (const context of this.#contexts.values()) {
context.initialCapture?.abort.abort();
context.settle.cancel();
}
}
buildTrace(input: {
drafts: RecordingDraftStep[];
startedAt: string;
purpose?: string;
startUrl?: string;
stoppedBy: StopReason;
bskVersion: string;
}): TraceV3 {
return buildTraceV3({
registry: this.#registry,
drafts: input.drafts,
annotations: this.#annotations,
startedAt: input.startedAt,
purpose: input.purpose,
startUrl: input.startUrl,
stoppedBy: input.stoppedBy,
bskVersion: input.bskVersion,
redactValues: this.#redactValues,
});
}
}
@@ -212,6 +212,7 @@ export class SettleController {
url: finalUrl,
pageUrl: finalUrl,
cause: "browser",
transitionQualifiers: ["server_redirect"],
preStateId: this.#session.cursor.lastSettled?.stateId,
};
drafts.push(draft);
@@ -2,7 +2,8 @@ export interface RecordedStateEntry {
id: string;
url: string;
title?: string;
rawVomText: string;
/** Rendered VOM text; capture indexes and source DOM/AX data never enter the registry. */
vomText: string;
truncated: boolean;
stepsHere: number[];
}
@@ -19,10 +20,10 @@ export class RecordingStateRegistry {
register(input: {
url: string;
title?: string;
rawVomText: string;
vomText: string;
truncated?: boolean;
}): RecordedStateEntry {
const identity = stateIdentity(input.url, input.rawVomText);
const identity = stateIdentity(input.url, input.vomText);
const existingId = this.#idByIdentity.get(identity);
if (existingId) {
const existing = this.#entriesById.get(existingId)!;
@@ -35,7 +36,7 @@ export class RecordingStateRegistry {
id: `s${this.#nextId}`,
url: input.url,
...(input.title ? { title: input.title } : {}),
rawVomText: input.rawVomText,
vomText: input.vomText,
truncated: input.truncated ?? false,
stepsHere: [],
};
+62 -48
View File
@@ -1,5 +1,6 @@
import type { RecordStepPayload } from "../record-bridge";
import type { RecordingDraftStep } from "./types";
import { hasRedirectQualifier } from "./navigation-policy";
import type { RecordingDraftStep, TargetMatchHint } from "./types";
export interface RecordingStepBuffer {
steps: RecordingDraftStep[];
@@ -10,33 +11,29 @@ export interface RecordingStepBuffer {
const NAVIGATION_TRIGGER_WINDOW_MS = 3_000;
function toDraftStep(payload: RecordStepPayload): RecordingDraftStep | null {
function toDraftStep(
payload: RecordStepPayload,
targetHint?: TargetMatchHint,
): RecordingDraftStep | null {
const pageUrl = payload.page_url;
const common = {
...(pageUrl ? { pageUrl } : {}),
...(targetHint ? { targetHint } : {}),
};
switch (payload.op) {
case "click":
return payload.target
? {
op: "click",
captureTarget: payload.target,
...(pageUrl ? { pageUrl } : {}),
}
: null;
return payload.target ? { op: "click", captureTarget: payload.target, ...common } : null;
case "hover":
return payload.target
? {
op: "hover",
captureTarget: payload.target,
...(pageUrl ? { pageUrl } : {}),
}
: null;
return payload.target ? { op: "hover", captureTarget: payload.target, ...common } : null;
case "fill":
return payload.target
? {
op: "fill",
captureTarget: payload.target,
value: payload.value ?? "",
...(payload.commit ? { commit: payload.commit } : {}),
...(payload.redacted ? { redacted: true } : {}),
...(pageUrl ? { pageUrl } : {}),
...common,
}
: null;
case "press":
@@ -46,7 +43,7 @@ function toDraftStep(payload: RecordStepPayload): RecordingDraftStep | null {
key: payload.key,
...(payload.target ? { captureTarget: payload.target } : {}),
...(payload.modifiers?.length ? { modifiers: payload.modifiers } : {}),
...(pageUrl ? { pageUrl } : {}),
...common,
}
: null;
case "select":
@@ -56,7 +53,7 @@ function toDraftStep(payload: RecordStepPayload): RecordingDraftStep | null {
captureTarget: payload.target,
values: payload.values,
...(payload.labels?.length ? { labels: payload.labels } : {}),
...(pageUrl ? { pageUrl } : {}),
...common,
}
: null;
case "navigate":
@@ -64,26 +61,35 @@ function toDraftStep(payload: RecordStepPayload): RecordingDraftStep | null {
}
}
function annotateLastStepNavigation(buffer: RecordingStepBuffer, url: string): boolean {
for (let i = buffer.steps.length - 1; i >= 0; i -= 1) {
const step = buffer.steps[i];
function annotateLastStepNavigation(buffer: RecordingStepBuffer, url: string): number {
for (let index = buffer.steps.length - 1; index >= 0; index -= 1) {
const step = buffer.steps[index];
if (!step) continue;
if (step.op === "click" || step.op === "press" || step.op === "select") {
buffer.steps[i] = { ...step, navigatedTo: url };
return true;
if (step.op === "click" || step.op === "press" || step.op === "select" || step.op === "fill") {
step.navigatedTo = url;
return index;
}
break;
}
return false;
return -1;
}
export type NavigationObserveResult =
| { kind: "noop" }
| { kind: "annotated"; index: number }
| { kind: "appended"; index: number }
| { kind: "coalesce_redirect"; url: string };
export function observeRecordedNavigation(
buffer: RecordingStepBuffer,
url: string,
causedByAction?: boolean,
): void {
if (!url || url === buffer.currentUrl) return;
transitionType?: string,
transitionQualifiers?: string[],
): NavigationObserveResult {
if (!url || url === buffer.currentUrl) return { kind: "noop" };
buffer.currentUrl = url;
const pendingIsCurrent =
buffer.pendingNavigation &&
(buffer.pendingNavigationDeadline === undefined ||
@@ -92,40 +98,47 @@ export function observeRecordedNavigation(
if (causedByAction === true || (causedByAction === undefined && pendingIsCurrent)) {
buffer.pendingNavigation = false;
buffer.pendingNavigationDeadline = undefined;
if (!annotateLastStepNavigation(buffer, url)) {
buffer.steps.push({
op: "navigate",
url,
pageUrl: url,
});
}
return;
const annotatedIndex = annotateLastStepNavigation(buffer, url);
if (annotatedIndex >= 0) return { kind: "annotated", index: annotatedIndex };
} else {
buffer.pendingNavigation = false;
buffer.pendingNavigationDeadline = undefined;
}
buffer.pendingNavigation = false;
buffer.pendingNavigationDeadline = undefined;
if (hasRedirectQualifier(transitionQualifiers)) {
return { kind: "coalesce_redirect", url };
}
buffer.steps.push({
op: "navigate",
url,
pageUrl: url,
transitionType,
transitionQualifiers,
});
return { kind: "appended", index: buffer.steps.length - 1 };
}
export function appendRecordedPayload(
buffer: RecordingStepBuffer,
payload: RecordStepPayload,
): void {
targetHint?: TargetMatchHint,
): number | null {
if (payload.op === "navigate") {
if (payload.url) {
observeRecordedNavigation(buffer, payload.url, payload.navigation_caused_by_action);
}
return;
if (!payload.url) return null;
const result = observeRecordedNavigation(
buffer,
payload.url,
payload.navigation_caused_by_action,
payload.transitionType,
payload.transitionQualifiers,
);
return result.kind === "appended" ? result.index : null;
}
const step = toDraftStep({
...payload,
page_url: payload.page_url ?? buffer.currentUrl,
});
if (!step) return;
const step = toDraftStep(
{ ...payload, page_url: payload.page_url ?? buffer.currentUrl },
targetHint,
);
if (!step) return null;
buffer.steps.push(step);
if (step.op === "click" || step.op === "press" || step.op === "select" || step.op === "fill") {
buffer.pendingNavigation = payload.expects_navigation === true;
@@ -133,4 +146,5 @@ export function appendRecordedPayload(
? Date.now() + NAVIGATION_TRIGGER_WINDOW_MS
: undefined;
}
return buffer.steps.length - 1;
}
@@ -39,8 +39,9 @@ export function buildTraceV3(input: {
startUrl?: string;
stoppedBy: StopReason;
bskVersion: string;
redactValues?: boolean;
}): TraceV3 {
const reduced = reduceTraceStepsV3(input.drafts);
const reduced = reduceTraceStepsV3(input.drafts, { redactValues: input.redactValues });
const entries = publishedEntries(input.registry, reduced.steps);
const publishedId = new Map(entries.map((entry, index) => [entry.id, `s${index + 1}`]));
const annotationsByState = new Map<string, StepAnnotation[]>();
@@ -65,7 +66,7 @@ export function buildTraceV3(input: {
url: entry.url,
title: entry.title,
stepIds: remapDraftIds(entry.stepsHere, reduced.stepIdByDraftId),
vomText: entry.rawVomText,
vomText: entry.vomText,
annotations: annotationsByState.get(entry.id) ?? [],
stepIdByDraftId: reduced.stepIdByDraftId,
}),
@@ -1,5 +1,6 @@
import type { NavigationCause, StepV3 } from "@/transport/types";
import { shouldIncludeDraft } from "./draft-policy";
import { hasRedirectQualifier } from "./navigation-policy";
import { unmatchedTarget } from "./target-matcher";
import type { RecordingDraftStep } from "./types";
@@ -8,7 +9,6 @@ interface CollapsedDraft {
draftIds: number[];
}
const REDIRECT_QUALIFIERS = new Set(["client_redirect", "server_redirect"]);
const TRANSITION_CAUSES: Record<string, NavigationCause> = {
typed: "user_typed",
generated: "user_typed",
@@ -22,7 +22,7 @@ const TRANSITION_CAUSES: Record<string, NavigationCause> = {
};
function isRedirect(step: Extract<RecordingDraftStep, { op: "navigate" }>): boolean {
return (step.transitionQualifiers ?? []).some((qualifier) => REDIRECT_QUALIFIERS.has(qualifier));
return hasRedirectQualifier(step.transitionQualifiers);
}
function collapseRedirects(steps: RecordingDraftStep[]): CollapsedDraft[] {
@@ -58,7 +58,7 @@ function selection(values: string[], labels?: string[]): Array<{ value: string;
}));
}
function reduceDraft(draft: RecordingDraftStep, id: number): StepV3 | null {
function reduceDraft(draft: RecordingDraftStep, id: number, redactValues: boolean): StepV3 | null {
if (!shouldIncludeDraft(draft)) return null;
const state = draft.preStateId ?? draft.postStateId;
const resultState = draft.postStateId ?? draft.preStateId;
@@ -81,13 +81,14 @@ function reduceDraft(draft: RecordingDraftStep, id: number): StepV3 | null {
target: draft.matchedTarget ?? unmatchedTarget(draft.captureTarget),
};
case "fill":
const fillIsRedacted = redactValues || draft.redacted === true;
return {
op: "fill",
...common,
target: draft.matchedTarget ?? unmatchedTarget(draft.captureTarget),
value: draft.value,
value: fillIsRedacted ? "***" : draft.value,
commit: draft.commit ?? "blur",
...(draft.redacted ? { redacted: true } : {}),
...(fillIsRedacted ? { redacted: true } : {}),
};
case "press":
return {
@@ -104,7 +105,7 @@ function reduceDraft(draft: RecordingDraftStep, id: number): StepV3 | null {
op: "select",
...common,
target: draft.matchedTarget ?? unmatchedTarget(draft.captureTarget),
selection: selection(draft.values, draft.labels),
...(!redactValues ? { selection: selection(draft.values, draft.labels) } : {}),
};
case "scroll":
return { op: "scroll", ...common };
@@ -116,11 +117,14 @@ export interface ReducedTraceV3 {
stepIdByDraftId: Map<number, number>;
}
export function reduceTraceStepsV3(steps: RecordingDraftStep[]): ReducedTraceV3 {
export function reduceTraceStepsV3(
steps: RecordingDraftStep[],
options: { redactValues?: boolean } = {},
): ReducedTraceV3 {
const output: StepV3[] = [];
const stepIdByDraftId = new Map<number, number>();
for (const { draft, draftIds } of collapseRedirects(steps)) {
const step = reduceDraft(draft, output.length + 1);
const step = reduceDraft(draft, output.length + 1, options.redactValues ?? false);
if (!step) continue;
output.push(step);
for (const draftId of draftIds) stepIdByDraftId.set(draftId, step.id);
File diff suppressed because it is too large Load Diff
+253 -37
View File
@@ -15,22 +15,27 @@ import {
type RecordQueryResponse,
type RecordStartAck,
type RecordStartMessage,
type RecordStepAck,
type RecordStopMessage,
} from "@/lib/record-bridge";
import { RecordingObservationRuntime } from "@/lib/recording/recording-runtime";
import { appendRecordedPayload, observeRecordedNavigation } from "@/lib/recording/step-buffer";
import { buildTraceV2 } from "@/lib/recording/trace-reducer-v2";
import type { RecordingDraftStep } from "@/lib/recording/types";
import type { SessionManager } from "@/session-manager/manager";
import { EXTENSION_VERSION } from "@/transport/handshake";
import type {
RecordAwaitParams,
RecordAwaitResult,
RecordedTrace,
RecordStartParams,
RecordStartResult,
RecordStopParams,
RecordStopResult,
RpcError,
TraceV2,
StopReason,
} from "@/transport/types";
import { TRACE_VERSION_V3 } from "@/transport/types";
import { handleNavigate } from "./navigation";
import {
type CdpRunner,
@@ -50,14 +55,32 @@ interface ActiveRecording {
steps: RecordingDraftStep[];
startedAt: string;
startedAtMs: number;
finishPromise: Promise<TraceV2>;
resolveFinish: (trace: TraceV2) => void;
traceVersion: 2 | 3;
finishPromise: Promise<RecordedTrace>;
resolveFinish: (trace: RecordedTrace) => void;
rejectFinish: (err: Error) => void;
settled: boolean;
finishing: boolean;
finishAttempt: Promise<RecordedTrace | null> | null;
currentUrl?: string;
pendingNavigation: boolean;
pendingNavigationDeadline?: number;
observation: RecordingObservationRuntime | null;
stoppedBy: StopReason;
/** Navigation callbacks tracked from event receipt through action enqueue. */
navigationCallbacks: Set<Promise<void>>;
/** Synchronous intake gate closed only after finish drains to stability. */
acceptingNavigation: boolean;
/**
* Serializes step appends with navigation observation so a click is always
* in `steps` before a same-turn `webNavigation` tries to annotate it.
*/
actionQueue: Promise<void>;
/** Last accepted sequence for each content-script document producer. */
lastStepSequenceByProducer: Map<string, number>;
}
function enqueueRecordingAction(recording: ActiveRecording, task: () => Promise<void>): void {
recording.actionQueue = recording.actionQueue.then(task, task).catch(() => {});
}
const recordings = new Map<string, ActiveRecording>();
@@ -78,6 +101,9 @@ function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/** Recording producer version mirrored into trace.recorder.bsk. */
export const BSK_TRACE_VERSION = EXTENSION_VERSION;
/** Injectable http(s) landing page when `tool.record_start` omits `url`. */
export const RECORD_DEFAULT_START_URL = "https://example.com/";
@@ -154,7 +180,29 @@ async function sendRecordStartWithAck(
throw lastError ?? new Error("failed to start recording in content script");
}
function buildTrace(recording: ActiveRecording): TraceV2 {
function negotiatedTraceVersion(params: RecordStartParams): 2 | 3 | RpcError {
if (params.trace_version === undefined) return 2;
if (params.trace_version === TRACE_VERSION_V3) return 3;
return {
code: "invalid_params",
message: `unsupported trace_version ${params.trace_version}; supported values are omitted (v2) or ${TRACE_VERSION_V3} (v3)`,
};
}
function buildTrace(recording: ActiveRecording): RecordedTrace {
if (recording.traceVersion === 3 && recording.observation) {
return recording.observation.buildTrace({
drafts: recording.steps,
startedAt: recording.startedAt,
purpose: recording.purpose,
startUrl: recording.startUrl,
stoppedBy: recording.stoppedBy,
bskVersion: BSK_TRACE_VERSION,
});
}
if (recording.traceVersion === 3) {
throw new Error("trace v3 observation runtime is unavailable");
}
return buildTraceV2({
steps: recording.steps,
startedAt: recording.startedAt,
@@ -163,6 +211,19 @@ function buildTrace(recording: ActiveRecording): TraceV2 {
});
}
async function processRecordedStep(
recording: ActiveRecording,
draftIndex: number,
tabId: number,
): Promise<void> {
if (!recording.observation) return;
try {
await recording.observation.processDraft(tabId, recording.steps, draftIndex);
} catch (err) {
console.warn(`[bsk record] observation failed for step ${draftIndex + 1}`, err);
}
}
export interface RecordDeps {
tabsApi: ChromeTabsApi;
sendToTab(
@@ -237,16 +298,41 @@ export function releaseBrowserObservationListenersIfIdle(): void {
detachBrowserObservation = null;
}
export function attachRecordStepListener(): () => void {
export function attachRecordStepListener(deps: RecordDeps = getDefaultDeps()): () => void {
const listener = (
message: unknown,
_sender: chrome.runtime.MessageSender,
_sendResponse: () => void,
sender: chrome.runtime.MessageSender,
sendResponse: (response: RecordStepAck) => void,
) => {
if (!isRecordStepMessage(message)) return false;
for (const recording of recordings.values()) {
if (recording.requestId !== message.requestId) continue;
appendRecordedPayload(recording, message.step);
const sourceTabId = sender.tab?.id ?? recording.tabId;
const producerKey = `${sourceTabId}:${message.producerId}`;
const expectedSequence = (recording.lastStepSequenceByProducer.get(producerKey) ?? 0) + 1;
if (message.sequence < expectedSequence) {
sendResponse({ ok: true, sequence: message.sequence });
return false;
}
if (message.sequence > expectedSequence) {
sendResponse({
ok: false,
expectedSequence,
error: `out-of-order recorded step ${message.sequence}`,
});
return false;
}
recording.lastStepSequenceByProducer.set(producerKey, message.sequence);
enqueueRecordingAction(recording, async () => {
await recording.observation?.flushRedirects();
const targetHint = message.step.geometry ? { geometry: message.step.geometry } : undefined;
const draftIndex = appendRecordedPayload(recording, message.step, targetHint);
if (draftIndex !== null) {
await processRecordedStep(recording, draftIndex, sourceTabId);
}
});
sendResponse({ ok: true, sequence: message.sequence });
return false;
}
return false;
@@ -428,11 +514,60 @@ export function attachRecordTabListener(deps: RecordDeps = getDefaultDeps()): ()
}
export function attachRecordNavigationListener(deps: RecordDeps = getDefaultDeps()): () => void {
const observeMainFrame = (tabId: number, url?: string, causedByAction?: boolean) => {
const observeMainFrame = (
tabId: number,
url?: string,
causedByAction?: boolean,
transitionType?: string,
transitionQualifiers?: string[],
) => {
if (!url) return;
const direct = findRecordingByTabId(tabId);
const candidates = direct
? direct.acceptingNavigation
? [direct]
: []
: [...recordings.values()].filter(
(recording) => !recording.settled && recording.acceptingNavigation,
);
if (candidates.length === 0) return;
let resolveTracked!: () => void;
const tracked = new Promise<void>((resolve) => {
resolveTracked = resolve;
});
for (const candidate of candidates) candidate.navigationCallbacks.add(tracked);
void (async () => {
const recording = await findRecordingForTab(tabId, deps);
if (recording && url) {
observeRecordedNavigation(recording, url, causedByAction);
try {
const recording = await findRecordingForTab(tabId, deps);
if (!recording || !recording.acceptingNavigation || !candidates.includes(recording)) {
return;
}
enqueueRecordingAction(recording, async () => {
const result = observeRecordedNavigation(
recording,
url,
causedByAction,
transitionType,
transitionQualifiers,
);
if (result.kind === "coalesce_redirect") {
recording.observation?.scheduleRedirect(tabId, recording.steps, result.url);
return;
}
if (result.kind === "noop") return;
recording.observation?.clearRedirect(tabId);
if (result.kind === "appended") {
await processRecordedStep(recording, result.index, tabId);
} else {
recording.observation?.scheduleSettle(tabId, recording.steps, result.index);
}
});
} finally {
for (const candidate of candidates) candidate.navigationCallbacks.delete(tracked);
resolveTracked();
}
})();
};
@@ -454,18 +589,13 @@ export function attachRecordNavigationListener(deps: RecordDeps = getDefaultDeps
details: chrome.webNavigation.WebNavigationTransitionCallbackDetails,
) => {
if (details.frameId !== 0) return;
const causedByAction =
details.transitionType === "link" || details.transitionType === "form_submit"
? true
: details.transitionType === "typed" ||
details.transitionType === "auto_bookmark" ||
details.transitionType === "generated" ||
details.transitionType === "keyword" ||
details.transitionType === "keyword_generated" ||
details.transitionType === "reload"
? false
: undefined;
observeMainFrame(details.tabId, details.url, causedByAction);
observeMainFrame(
details.tabId,
details.url,
undefined,
details.transitionType,
details.transitionQualifiers,
);
};
chrome.webNavigation.onCompleted.addListener(completedListener);
chrome.webNavigation.onCommitted?.addListener(committedListener);
@@ -483,6 +613,26 @@ export function attachRecordNavigationListener(deps: RecordDeps = getDefaultDeps
return () => chrome.tabs.onUpdated.removeListener(listener);
}
const MAX_FINISH_DRAIN_ROUNDS = 10;
async function drainRecordingToStability(recording: ActiveRecording): Promise<boolean> {
for (let round = 0; round < MAX_FINISH_DRAIN_ROUNDS; round += 1) {
await Promise.all([...recording.navigationCallbacks]);
const actionTail = recording.actionQueue;
await actionTail;
await recording.observation?.flush();
if (recording.navigationCallbacks.size === 0 && recording.actionQueue === actionTail) {
// No event or promise continuation can interleave with this synchronous
// check-and-close, so work accepted before the cutoff is fully drained.
recording.acceptingNavigation = false;
return true;
}
}
console.warn("[bsk record] navigation/action queues did not stabilize at stop");
return false;
}
export function attachRecordQueryListener(deps: RecordDeps = getDefaultDeps()): () => void {
const listener = (
message: unknown,
@@ -523,23 +673,50 @@ async function finishRecordingByRequest(
if (recording.requestId !== requestId || recording.settled) continue;
const match = await findRecordingForTab(tabId, deps);
if (match !== recording) continue;
await finishRecording(sessionId, deps);
await finishRecording(sessionId, deps, "user_finish");
return;
}
}
async function finishRecording(sessionId: string, deps: RecordDeps): Promise<TraceV2 | null> {
async function finishRecording(
sessionId: string,
deps: RecordDeps,
stoppedBy: StopReason,
): Promise<RecordedTrace | null> {
const recording = recordings.get(sessionId);
if (!recording || recording.settled || recording.finishing) return null;
recording.finishing = true;
if (!recording || recording.settled) return null;
if (recording.finishAttempt) return recording.finishAttempt;
recording.stoppedBy = stoppedBy;
const attempt = finishRecordingAttempt(sessionId, recording, deps);
recording.finishAttempt = attempt;
try {
return await attempt;
} finally {
if (!recording.settled) {
recording.finishAttempt = null;
}
}
}
async function finishRecordingAttempt(
sessionId: string,
recording: ActiveRecording,
deps: RecordDeps,
): Promise<RecordedTrace | null> {
await clearRearmTimersForRecording(recording, deps);
try {
// Disposing content capture may commit a final dirty fill. Flush content
// first so all resulting RECORD_STEP messages enter actionQueue before it
// and the observation queues are drained.
await stopRecordingOnAllAgentTabs(recording, deps);
} catch {
recording.finishing = false;
return null;
}
if (!(await drainRecordingToStability(recording))) {
return null;
}
await recording.observation?.settleTrailing(recording.tabId, recording.steps);
recording.settled = true;
recordings.delete(sessionId);
@@ -554,6 +731,15 @@ export async function handleRecordStart(
params: RecordStartParams,
deps: RecordDeps = getDefaultDeps(),
): Promise<RecordStartResult | RpcError> {
const traceVersionOrErr = negotiatedTraceVersion(params);
if (typeof traceVersionOrErr !== "number") return traceVersionOrErr;
if (traceVersionOrErr === 3 && !deps.cdp) {
return {
code: "protocol_error",
message: "trace v3 recording requires an active CDP connection",
};
}
const ctxOrErr = lookupSession(manager, params, "record_start");
if (isRpcError(ctxOrErr)) return ctxOrErr;
const ctx = ctxOrErr;
@@ -570,14 +756,16 @@ export async function handleRecordStart(
// on the destination page can RECORD_QUERY → rearm → show RecordOverlay
// instead of flashing ControlOverlay ("Agent 正在控制").
const requestId = makeRequestId(target.tabId);
let resolveFinish!: (trace: TraceV2) => void;
let resolveFinish!: (trace: RecordedTrace) => void;
let rejectFinish!: (err: Error) => void;
const finishPromise = new Promise<TraceV2>((resolve, reject) => {
const finishPromise = new Promise<RecordedTrace>((resolve, reject) => {
resolveFinish = resolve;
rejectFinish = reject;
});
const navigateUrl = params.url ?? RECORD_DEFAULT_START_URL;
const startedAtMs = Date.now();
const maxPageTokens = params.max_page_tokens;
const redactValues = params.redact_values ?? false;
recordings.set(params.session_id, {
requestId,
tabId: target.tabId,
@@ -587,20 +775,36 @@ export async function handleRecordStart(
steps: [],
startedAt: new Date(startedAtMs).toISOString(),
startedAtMs,
traceVersion: traceVersionOrErr,
finishPromise,
resolveFinish,
rejectFinish,
settled: false,
finishing: false,
finishAttempt: null,
currentUrl: navigateUrl,
pendingNavigation: false,
pendingNavigationDeadline: undefined,
observation:
traceVersionOrErr === 3 && deps.cdp
? new RecordingObservationRuntime({
cdp: deps.cdp,
tabsApi: deps.tabsApi,
maxTokens: maxPageTokens,
redactValues,
})
: null,
stoppedBy: "user_finish",
navigationCallbacks: new Set(),
acceptingNavigation: true,
actionQueue: Promise.resolve(),
lastStepSequenceByProducer: new Map(),
});
// Observe navigations for the whole recording lifetime; attach before
// optional navigate so the destination load can rearm capture.
ensureBrowserObservationListeners(deps);
const abortPending = async (notifyContent: boolean) => {
recordings.get(params.session_id)?.observation?.cancel();
recordings.delete(params.session_id);
releaseBrowserObservationListenersIfIdle();
if (notifyContent) {
@@ -690,7 +894,7 @@ export async function handleRecordStart(
code: "invalid_params",
message: params.url
? `cannot record on restricted URL (${startUrl}); use an http(s) page`
: `cannot record on restricted URL (${startUrl}); default start page must be injectable http(s)`,
: `cannot record on restricted URL (${startUrl}); default start page https://example.com/ did not load — pass --url with a page you can open`,
};
}
@@ -727,6 +931,17 @@ export async function handleRecordStart(
};
}
if (active.observation) {
const activeRecording = recordings.get(params.session_id);
if (activeRecording) {
try {
await activeRecording.observation?.captureInitial(target.tabId);
} catch {
// Proceed without initial observation; steps may be dropped by reducer.
}
}
}
{
const cancelled = await abortIfCancelled(true);
if (cancelled) return cancelled;
@@ -751,7 +966,7 @@ export async function handleRecordStop(
};
}
const trace = await finishRecording(params.session_id, deps);
const trace = await finishRecording(params.session_id, deps, "cli_stop");
if (!trace) {
return {
code: "protocol_error",
@@ -781,9 +996,9 @@ export async function handleRecordAwait(
return { code: "cancelled", message: "record_await aborted" };
}
const outcome = await new Promise<{ trace: TraceV2 } | { error: RpcError }>((resolve) => {
const outcome = await new Promise<{ trace: RecordedTrace } | { error: RpcError }>((resolve) => {
let settled = false;
const finish = (result: { trace: TraceV2 } | { error: RpcError }) => {
const finish = (result: { trace: RecordedTrace } | { error: RpcError }) => {
if (settled) return;
settled = true;
if (timer) clearTimeout(timer);
@@ -825,6 +1040,7 @@ export function clearRecordingForSession(sessionId: string): void {
}
void clearRearmTimersForRecording(recording, getDefaultDeps());
if (!recording.settled) {
recording.observation?.cancel();
recording.settled = true;
recording.rejectFinish(new Error("recording cleared"));
}
@@ -65,6 +65,11 @@ function deferredFakeTransport(): { transport: Transport; emit: (frame: Protocol
}
describe("performHandshake", () => {
it("advertises the protocol compatibility boundary", () => {
expect(PROTOCOL_VERSION).toBe("1.1");
expect(MIN_COMPATIBLE_PROTOCOL).toBe("1.0");
});
it("sends system.handshake with identity and both compat fields", async () => {
let sentFrame: ProtocolFrame | null = null;
const transport = fakeTransport((req) => {
@@ -74,7 +79,7 @@ describe("performHandshake", () => {
result: {
server: "browser-skill-daemon",
version: "0.1.0",
protocol_version: "1.0",
protocol_version: "1.1",
min_compatible_peer: "0.0.0",
min_compatible_protocol: "1.0",
},
@@ -129,8 +134,8 @@ describe("performHandshake", () => {
const response = {
server: "browser-skill-daemon",
version: "0.1.0",
protocol_version: "1.0",
min_compatible_protocol: "1.0",
protocol_version: "1.1",
min_compatible_protocol: "1.1",
} satisfies HandshakeResult;
const transport = fakeTransport((req) => ({
id: (req as { id: string }).id,
@@ -145,7 +150,7 @@ describe("performHandshake", () => {
});
expect(outcome.result.min_compatible_peer).toBeUndefined();
expect(outcome.result.min_compatible_protocol).toBe("1.0");
expect(outcome.result.min_compatible_protocol).toBe("1.1");
});
it("rejects when the daemon responds with an error", async () => {
@@ -178,9 +183,9 @@ describe("performHandshake", () => {
result: {
server: "browser-skill-daemon",
version: "0.1.0",
protocol_version: "1.0",
protocol_version: "1.1",
min_compatible_peer: "0.0.0",
min_compatible_protocol: "1.0",
min_compatible_protocol: "1.1",
},
});
emit({
@@ -188,14 +193,14 @@ describe("performHandshake", () => {
result: {
server: "browser-skill-daemon",
version: "0.1.0",
protocol_version: "1.0",
protocol_version: "1.1",
min_compatible_peer: "0.0.0",
min_compatible_protocol: "1.0",
min_compatible_protocol: "1.1",
},
});
await expect(pending).resolves.toMatchObject({
result: { server: "browser-skill-daemon", protocol_version: "1.0" },
result: { server: "browser-skill-daemon", protocol_version: "1.1" },
});
});
+1 -1
View File
@@ -7,7 +7,7 @@ import type {
ResponseFrame,
} from "./types";
export const PROTOCOL_VERSION = "1.0";
export const PROTOCOL_VERSION = "1.1";
/**
* Extension semver, injected at build time from `package.json` via
* Vite's `define` (see `wxt.config.ts` and `vitest.config.ts`).
+4
View File
@@ -819,6 +819,10 @@ export interface RecordStartParams {
tab_id?: number;
url?: string;
purpose?: string;
max_page_tokens?: number;
redact_values?: boolean;
/** Omitted means v2; `3` requests a state-linked v3 trace. */
trace_version?: number;
}
export interface RecordStartResult {
+14 -6
View File
@@ -265,19 +265,27 @@ retry — complete the task autonomously or stop gracefully.
### Recording — `bsk record`
Capture the user's own actions in the Agent Window to a `trace.json`, for later LLM-driven automation:
Capture the user's own actions in the Agent Window to a **trace bundle**, for later LLM-driven automation. New CLI builds request **trace v3** (page observations + action chain); older extensions may still return **trace v2** (actions only), which the CLI exports as a single `trace.json` without `states/`.
```bash
bsk record start --browser <instance-id-or-label> [--url https://…] [--purpose "publish a wiki doc"] [--output trace.json]
bsk record start --browser <instance-id-or-label> \
[--url https://…] [--purpose "publish a wiki doc"] \
[--max-page-tokens 3000] [--redact-values] \
[--output trace]
# `--url` is optional; default https://example.com/ when omitted (must be http(s)).
# Blocks until the user clicks Finish in the recording panel, then writes ./trace.json and closes the window.
# Blocks until the user clicks Finish in the recording panel, then writes:
# trace/trace.json — action chain (+ state index when v3)
# trace/states/ — v3 only: one `sN.txt` observe snapshot per settled page state
bsk record stop [--output trace.json] # terminal fallback if the browser panel is unavailable
bsk record stop [--output trace] # terminal fallback if the browser panel is unavailable
```
- The trace is a **record-only action log** (a `pages[]` dictionary + `navigate`/`click`/`fill`/`select`/`press` steps with `target` descriptors). It records *what the user did*; deciding which inputs are variable is left to the executing agent.
- **v3 bundle (preferred):** `--output` is a directory (default `./trace`) containing `trace.json` and `states/`. `trace.json` lists `states[]` (page observation ids) and `steps[]` (each step binds `state` = observe snapshot *before* the action and `result.state` = snapshot *after* settle). Page bodies live in `states/sN.txt` using the same VOM format as `bsk observe`.
- **v2 fallback:** when the connected extension is older, export may contain only legacy `trace.json` with `pages[]` action context and no `states/` observation files. Update the extension for full v3 bundles.
- Each `states[]` entry is one **settled page observation** (captured after recording start, navigation landing, or action settle — not on a timer).
- `target.ref` values like `@e12` exist **only inside** the bundle for disambiguation; do **not** copy `@eN` refs into SKILL.md or agent runbooks — use visible names from the observation text instead.
- `--purpose` is optional context metadata; it does **not** change what gets captured.
- There is **no** `bsk replay` — to redo a flow, read the trace and reuse the existing `session` / `snapshot` / `@eN` / `click` / `fill` tools. Follow **Stop when the goal is met**.
- `--redact-values` masks all form values in page files as `[filled]` / `[empty]`.
- Do **not** record on banking/SSO/password-manager pages; passwords are redacted but traces may still contain sensitive text.
## Error handling
+1
View File
@@ -22,6 +22,7 @@ pub mod navigate;
pub mod network;
pub mod observe;
pub mod record;
pub mod record_recovery;
pub mod record_state;
pub mod render_error;
pub mod screenshot;
+268 -71
View File
@@ -1,23 +1,30 @@
//! `bsk record start|stop` — capture user actions in the Agent Window.
use std::fs;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::Context;
use bsk_protocol::Method;
use bsk_protocol::tools::{
RecordAwaitParams, RecordAwaitResult, RecordStartParams, RecordStartResult, RecordStopParams,
RecordStopResult, RecordedTrace,
RecordStopResult, RecordedTrace, TRACE_VERSION_V3,
};
use bsk_protocol::{ErrorCode, Method};
use clap::{Args, Subcommand};
use crate::cli::TOOL_IPC_TIMEOUT;
mod export;
use crate::cli::business_rpc;
use crate::cli::ensure_daemon::ensure_daemon;
use crate::cli::error::{CliError, Format};
use crate::cli::record_recovery;
use crate::cli::record_state;
use crate::cli::session::{SessionStartOptions, start_session, stop_session};
use export::{
ExportMeta, export_with_recovery, states_dir_for_output, trace_json_path,
validate_record_output,
};
/// Max wait for the user to click 结束 in the browser (24 hours).
const RECORD_AWAIT_TIMEOUT_MS: u32 = 86_400_000;
@@ -50,7 +57,8 @@ pub struct RecordStartArgs {
pub tab_id: Option<i64>,
/// Navigate to this http(s) URL before recording. When omitted, defaults
/// to `https://example.com/`.
/// to `https://example.com/`. If that page does not load, pass `--url`
/// with a site you can open in this browser.
#[arg(long)]
pub url: Option<String>,
@@ -58,15 +66,25 @@ pub struct RecordStartArgs {
#[arg(long)]
pub purpose: Option<String>,
/// Output path for the trace JSON (default `./trace.json`).
#[arg(long, default_value = "trace.json")]
/// Max VOM tokens per page observation file (default 3000).
#[arg(long = "max-page-tokens")]
pub max_page_tokens: Option<u32>,
/// Redact all form values in page observations (`[filled]` only).
#[arg(long = "redact-values")]
pub redact_values: bool,
/// Output directory for the trace bundle (default `./trace`).
/// Writes `<dir>/trace.json` and `<dir>/states/*.txt`.
#[arg(long, default_value = "trace")]
pub output: PathBuf,
}
#[derive(Debug, Clone, Args)]
pub struct RecordStopArgs {
/// Output path for the trace JSON (default `./trace.json`).
#[arg(long, default_value = "trace.json")]
/// Output directory for the trace bundle (default `./trace`).
/// Writes `<dir>/trace.json` and `<dir>/states/*.txt`.
#[arg(long, default_value = "trace")]
pub output: PathBuf,
}
@@ -78,11 +96,7 @@ pub fn dispatch(cmd: RecordCmd, format: Format) -> Result<(), CliError> {
}
fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError> {
if record_state::read().is_ok() {
return Err(CliError::Local(anyhow::anyhow!(
"a recording is already in progress; run `bsk record stop` first"
)));
}
prepare_record_start(&args.output)?;
let info = ensure_daemon().context("ensure daemon is running")?;
let session = start_session(
@@ -96,8 +110,11 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
let start_params = RecordStartParams {
session_id: session.session_id.clone(),
tab_id: args.tab_id,
url: args.url,
url: args.url.clone(),
purpose: args.purpose.clone(),
max_page_tokens: args.max_page_tokens,
redact_values: Some(args.redact_values),
trace_version: Some(TRACE_VERSION_V3),
};
let start_result = business_rpc::call::<RecordStartParams, RecordStartResult>(
info.sock_path.clone(),
@@ -111,7 +128,7 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
Ok(result) => result,
Err(err) => {
let _ = stop_session(info.sock_path, &session.session_id);
return Err(err);
return Err(annotate_default_start_page_error(err, args.url.as_deref()));
}
};
@@ -142,8 +159,8 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
// Keep write/render inside a Result so `?` cannot skip session teardown.
let run_result: Result<(), CliError> = match await_result {
Ok(result) => (|| {
write_trace_file(&args.output, &result.trace)?;
render_finish(&result.trace, &args.output, format)
let exported = export_with_recovery(&args.output, &result.trace)?;
render_finish(&result.trace, &args.output, &exported, format)
})(),
Err(err) => Err(err),
};
@@ -157,11 +174,12 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
}
fn dispatch_stop(args: RecordStopArgs, format: Format) -> Result<(), CliError> {
let state = record_state::read().map_err(CliError::Local)?;
let info = ensure_daemon().context("ensure daemon is running")?;
let session_id = state.session_id.clone();
validate_record_output(&args.output)?;
if let Ok(state) = record_state::read() {
let info = ensure_daemon().context("ensure daemon is running")?;
let session_id = state.session_id.clone();
let run_result: Result<(), CliError> = (|| {
let params = RecordStopParams {
session_id: session_id.clone(),
};
@@ -172,16 +190,85 @@ fn dispatch_stop(args: RecordStopArgs, format: Format) -> Result<(), CliError> {
Some(params),
TOOL_IPC_TIMEOUT,
)?;
write_trace_file(&args.output, &result.trace)?;
render_stop(&result, &args.output, format)
})();
let session_stop_result = stop_session(info.sock_path, &session_id);
record_state::clear();
let run_result: Result<(), CliError> = (|| {
let exported = export_with_recovery(&args.output, &result.trace)?;
render_stop(&result, &args.output, &exported, format)
})();
run_result?;
session_stop_result?;
Ok(())
let session_stop_result = stop_session(info.sock_path, &session_id);
record_state::clear();
run_result?;
session_stop_result?;
return Ok(());
}
let Some(trace) = record_recovery::load().map_err(CliError::Local)? else {
return Err(CliError::Local(anyhow::anyhow!(
"no recording in progress; run `bsk record start` first"
)));
};
let exported = export_with_recovery(&args.output, &trace)?;
render_finish(&trace, &args.output, &exported, format)
}
/// When `record start` omitted `--url` and the default example.com page
/// never loaded, rewrite the RPC error so the CLI hint points at `--url`.
fn annotate_default_start_page_error(err: CliError, user_url: Option<&str>) -> CliError {
if user_url.is_some() {
return err;
}
let CliError::Rpc {
code,
message,
data,
source,
} = err
else {
return err;
};
if !is_default_start_page_failure(code, &message) {
return CliError::Rpc {
code,
message,
data,
source,
};
}
CliError::Rpc {
code,
message,
data: Some(serde_json::json!({
"reason": crate::cli::render_error::reason::RECORD_START_PAGE_UNREACHABLE
})),
source,
}
}
fn is_default_start_page_failure(code: ErrorCode, message: &str) -> bool {
match code {
ErrorCode::CdpFailed | ErrorCode::Timeout => true,
ErrorCode::InvalidParams => {
message.contains("restricted URL") || message.contains("about:blank")
}
_ => false,
}
}
fn prepare_record_start(output: &Path) -> Result<(), CliError> {
if record_state::read().is_ok() {
return Err(CliError::Local(anyhow::anyhow!(
"a recording is already in progress; run `bsk record stop` first"
)));
}
if record_recovery::exists() {
return Err(CliError::Local(anyhow::anyhow!(
"a previous recording was not exported; run `bsk record stop --output <dir>` to recover it first"
)));
}
validate_record_output(output)
}
fn record_await_ipc_timeout(timeout_ms: u32) -> Duration {
@@ -190,43 +277,66 @@ fn record_await_ipc_timeout(timeout_ms: u32) -> Duration {
.unwrap_or(Duration::from_secs(u64::from(timeout_ms / 1_000) + 15))
}
fn write_trace_file(output: &PathBuf, trace: &RecordedTrace) -> Result<(), CliError> {
let json = serde_json::to_string_pretty(trace)
.context("serialize trace JSON")
.map_err(CliError::Local)?;
if let Some(parent) = output.parent() {
if !parent.as_os_str().is_empty() {
fs::create_dir_all(parent)
.with_context(|| format!("create output directory {}", parent.display()))
.map_err(CliError::Local)?;
}
}
fs::write(output, format!("{json}\n"))
.with_context(|| format!("write trace to {}", output.display()))
.map_err(CliError::Local)?;
Ok(())
}
fn render_finish(trace: &RecordedTrace, output: &PathBuf, format: Format) -> Result<(), CliError> {
fn render_finish(
trace: &RecordedTrace,
output_dir: &PathBuf,
exported: &ExportMeta,
format: Format,
) -> Result<(), CliError> {
let trace_path = trace_json_path(output_dir);
match format {
Format::Json => {
let payload = match trace {
RecordedTrace::V3(t) => serde_json::json!({
"output": output_dir,
"trace_json": trace_path,
"trace_version": exported.trace_version,
"states_dir": exported.states_dir,
"trace": t,
"window_closed": true,
}),
RecordedTrace::V2(t) => serde_json::json!({
"output": output_dir,
"trace_json": trace_path,
"trace_version": exported.trace_version,
"states_dir": null,
"trace": t,
"window_closed": true,
}),
};
println!(
"{}",
serde_json::to_string(&serde_json::json!({
"output": output,
"trace": trace,
"window_closed": true,
}))
.map_err(|e| CliError::Local(anyhow::anyhow!(e)))?
serde_json::to_string(&payload).map_err(|e| CliError::Local(anyhow::anyhow!(e)))?
);
}
Format::Human => {
let step_count = match trace {
RecordedTrace::V2(trace) => trace.steps.len(),
RecordedTrace::V3(trace) => trace.steps.len(),
};
println!("saved {step_count} steps to {}", output.display());
}
Format::Human => match trace {
RecordedTrace::V3(t) => {
let states_dir = exported
.states_dir
.as_ref()
.map(|p| p.display().to_string())
.unwrap_or_else(|| states_dir_for_output(output_dir).display().to_string());
println!(
"saved {} steps to {} and {} states to {}",
t.steps.len(),
trace_path.display(),
t.states.len(),
states_dir
);
}
RecordedTrace::V2(t) => {
if exported.v2_fallback {
eprintln!(
"note: extension returned trace v2 (no page observations); update the BrowserSkill extension for v3 bundles"
);
}
println!(
"saved {} steps to {} (trace v2)",
t.steps.len(),
trace_path.display()
);
}
},
}
Ok(())
}
@@ -234,33 +344,89 @@ fn render_finish(trace: &RecordedTrace, output: &PathBuf, format: Format) -> Res
fn render_stop(
result: &RecordStopResult,
output: &PathBuf,
exported: &ExportMeta,
format: Format,
) -> Result<(), CliError> {
render_finish(&result.trace, output, format)
render_finish(&result.trace, output, exported, format)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn default_output_is_trace_json() {
let args = RecordStopArgs {
output: PathBuf::from("trace.json"),
};
assert_eq!(args.output, PathBuf::from("trace.json"));
fn with_temp_home<F: FnOnce()>(f: F) {
let _lock = record_recovery::test_env_lock();
let tmp = tempfile::tempdir().unwrap();
unsafe {
std::env::set_var(crate::daemon::paths::BSK_HOME_ENV, tmp.path());
}
f();
record_recovery::clear();
record_state::clear();
unsafe {
std::env::remove_var(crate::daemon::paths::BSK_HOME_ENV);
}
}
#[test]
fn start_args_default_output_is_trace_json() {
fn prepare_record_start_rejects_existing_trace_json_file() {
with_temp_home(|| {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("trace.json");
std::fs::write(&output, "{}\n").unwrap();
let err = prepare_record_start(&output).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("JSON file") || msg.contains("json file"),
"{msg}"
);
assert!(
msg.contains("states/") || msg.contains("--output trace"),
"{msg}"
);
});
}
#[test]
fn prepare_record_start_rejects_unexported_recovery() {
with_temp_home(|| {
crate::daemon::paths::ensure_bsk_home().unwrap();
std::fs::write(
crate::daemon::paths::record_recovery_path().unwrap(),
"{}\n",
)
.unwrap();
let err = prepare_record_start(Path::new("trace")).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("not exported") || msg.contains("recover"),
"{msg}"
);
});
}
#[test]
fn default_output_is_trace_dir() {
let args = RecordStopArgs {
output: PathBuf::from("trace"),
};
assert_eq!(args.output, PathBuf::from("trace"));
}
#[test]
fn start_args_default_output_is_trace_dir() {
let args = RecordStartArgs {
browser: None,
tab_id: None,
url: None,
purpose: None,
output: PathBuf::from("trace.json"),
max_page_tokens: None,
redact_values: false,
output: PathBuf::from("trace"),
};
assert_eq!(args.output, PathBuf::from("trace.json"));
assert_eq!(args.output, PathBuf::from("trace"));
}
#[test]
@@ -268,4 +434,35 @@ mod tests {
let got = record_await_ipc_timeout(RECORD_AWAIT_TIMEOUT_MS);
assert!(got >= Duration::from_secs(86_400));
}
#[test]
fn annotate_default_start_page_error_adds_url_reason() {
let err = CliError::from_rpc(bsk_protocol::RpcError {
code: ErrorCode::CdpFailed,
message: "Page.navigate rejected: net::ERR_ABORTED".into(),
data: None,
});
let annotated = annotate_default_start_page_error(err, None);
assert_eq!(
crate::cli::render_error::reason_for_data(annotated.data()),
Some(crate::cli::render_error::reason::RECORD_START_PAGE_UNREACHABLE)
);
let info =
crate::cli::render_error::info_for_error(annotated.code().unwrap(), annotated.data());
assert!(info.hint.unwrap().contains("--url"));
}
#[test]
fn annotate_default_start_page_error_skips_when_user_passed_url() {
let err = CliError::from_rpc(bsk_protocol::RpcError {
code: ErrorCode::CdpFailed,
message: "Page.navigate rejected: net::ERR_ABORTED".into(),
data: None,
});
let annotated = annotate_default_start_page_error(err, Some("https://www.example.org/"));
assert_eq!(
crate::cli::render_error::reason_for_data(annotated.data()),
None
);
}
}
+976
View File
@@ -0,0 +1,976 @@
use std::collections::HashSet;
use std::fs::{self, OpenOptions};
use std::io;
use std::path::{Path, PathBuf};
use anyhow::Context;
use bsk_protocol::tools::{RecordedTrace, TRACE_VERSION_V2, TRACE_VERSION_V3, TraceV2, TraceV3};
use serde::Serialize;
use crate::cli::error::CliError;
#[derive(Debug)]
pub(super) struct ExportMeta {
pub(super) states_dir: Option<PathBuf>,
pub(super) trace_version: u32,
/// True when the CLI requested v3 but the extension returned legacy v2.
pub(super) v2_fallback: bool,
}
pub(super) fn export_recorded_trace(
output_dir: &Path,
trace: &RecordedTrace,
) -> Result<ExportMeta, CliError> {
match trace {
RecordedTrace::V3(trace) => {
let states_dir = write_trace_bundle(output_dir, trace)?;
Ok(ExportMeta {
states_dir: Some(states_dir),
trace_version: TRACE_VERSION_V3,
v2_fallback: false,
})
}
RecordedTrace::V2(trace) => {
write_trace_v2(output_dir, trace)?;
Ok(ExportMeta {
states_dir: None,
trace_version: TRACE_VERSION_V2,
v2_fallback: true,
})
}
}
}
/// Save the completed Trace first, then write the bundle. Recovery is
/// removed only after the export commits so a bad `--output` cannot drop
/// a recording the extension already returned.
pub(super) fn export_with_recovery(
output_dir: &Path,
trace: &RecordedTrace,
) -> Result<ExportMeta, CliError> {
let save_result = crate::cli::record_recovery::save(trace);
match export_recorded_trace(output_dir, trace) {
Ok(meta) => {
crate::cli::record_recovery::clear();
Ok(meta)
}
Err(export_err) => {
let export_err = annotate_recovery_hint(export_err, save_result.is_ok());
if let Err(save_err) = save_result {
return Err(match export_err {
CliError::Local(inner) => CliError::Local(
inner.context(format!("also failed to save recovery data: {save_err}")),
),
other => other,
});
}
Err(export_err)
}
}
}
fn annotate_recovery_hint(err: CliError, recovery_saved: bool) -> CliError {
if !recovery_saved {
return err;
}
match err {
CliError::Local(inner) => CliError::Local(inner.context(
"export failed; the recorded trace was saved and can be recovered with `bsk record stop --output <dir>`",
)),
other => other,
}
}
fn looks_like_json_output(path: &Path) -> bool {
path.extension()
.and_then(|ext| ext.to_str())
.is_some_and(|ext| ext.eq_ignore_ascii_case("json"))
}
fn legacy_json_output_error(path: &Path) -> CliError {
CliError::Local(anyhow::anyhow!(
"--output {} is a JSON file path; Trace v3 writes a bundle directory \
(`<dir>/trace.json` and `<dir>/states/`), not a single file. \
Use `--output trace` or another directory.",
path.display()
))
}
/// Reject legacy `--output trace.json` (and any existing non-directory)
/// before a recording starts, so the user is not blocked only after Finish.
pub(super) fn validate_record_output(path: &Path) -> Result<(), CliError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_symlink() => Err(CliError::Local(anyhow::anyhow!(
"--output {} must not be a symlink",
path.display()
))),
Ok(metadata) if metadata.is_dir() => Ok(()),
Ok(_) if looks_like_json_output(path) => Err(legacy_json_output_error(path)),
Ok(_) => Err(CliError::Local(anyhow::anyhow!(
"--output {} is not a directory; Trace v3 writes a bundle directory \
(`<dir>/trace.json` and `<dir>/states/`). Use `--output trace`.",
path.display()
))),
Err(err) if err.kind() == io::ErrorKind::NotFound => {
if looks_like_json_output(path) {
Err(legacy_json_output_error(path))
} else {
Ok(())
}
}
Err(err) => Err(CliError::Local(
anyhow::Error::new(err).context(format!("inspect --output {}", path.display())),
)),
}
}
fn acquire_export_lock(output_dir: &Path) -> Result<std::fs::File, CliError> {
validate_bundle_directory(output_dir, "output bundle directory")?;
fs::create_dir_all(output_dir)
.with_context(|| format!("create output dir {}", output_dir.display()))
.map_err(CliError::Local)?;
let lock_path = output_dir.join(".bsk-record-export.lock");
let lock_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&lock_path)
.with_context(|| format!("open export lock {}", lock_path.display()))
.map_err(CliError::Local)?;
fs2::FileExt::lock_exclusive(&lock_file)
.with_context(|| format!("lock export bundle {}", output_dir.display()))
.map_err(CliError::Local)?;
Ok(lock_file)
}
pub(super) fn write_trace_v2(output_dir: &Path, trace: &TraceV2) -> Result<(), CliError> {
let trace_path = trace_json_path(output_dir);
let states_dir = states_dir_for_output(output_dir);
let json = serde_json::to_string_pretty(trace)
.context("serialize trace JSON")
.map_err(CliError::Local)?;
let _lock = acquire_export_lock(output_dir)?;
validate_replaceable_file(&trace_path, "trace JSON")?;
let transaction_id = uuid::Uuid::new_v4();
let staging_dir = output_dir.join(format!(".bsk-record-stage-{transaction_id}"));
fs::create_dir_all(&staging_dir)
.with_context(|| format!("create staging dir {}", staging_dir.display()))
.map_err(CliError::Local)?;
let staged_trace = staging_dir.join("trace.json");
let stage_result: Result<(), CliError> = (|| {
fs::write(&staged_trace, format!("{json}\n"))
.with_context(|| format!("write staged trace to {}", staged_trace.display()))
.map_err(CliError::Local)?;
Ok(())
})();
if let Err(err) = stage_result {
let _ = fs::remove_dir_all(&staging_dir);
return Err(err);
}
commit_export_transaction(
&staging_dir,
&states_dir,
&HashSet::new(),
transaction_id,
vec![(
staged_trace,
trace_path,
output_dir.join(format!(".trace.json.{transaction_id}.bak")),
)],
)?;
Ok(())
}
pub(super) fn states_dir_for_output(output_dir: &Path) -> PathBuf {
output_dir.join("states")
}
pub(super) fn trace_json_path(output_dir: &Path) -> PathBuf {
output_dir.join("trace.json")
}
fn is_canonical_state_id(id: &str) -> bool {
let Some(number) = id.strip_prefix('s') else {
return false;
};
let mut digits = number.bytes();
matches!(digits.next(), Some(b'1'..=b'9')) && digits.all(|byte| byte.is_ascii_digit())
}
fn validate_bundle_directory(path: &Path, description: &str) -> Result<(), CliError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_symlink() => Err(CliError::Local(anyhow::anyhow!(
"{description} {} must not be a symlink",
path.display()
))),
Ok(metadata) if metadata.is_dir() => Ok(()),
Ok(_) if looks_like_json_output(path) => Err(legacy_json_output_error(path)),
Ok(_) => Err(CliError::Local(anyhow::anyhow!(
"{description} {} is not a directory; Trace v3 writes `<dir>/trace.json` and `<dir>/states/`. Use `--output trace`.",
path.display()
))),
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(CliError::Local(
anyhow::Error::new(err).context(format!("inspect {description} {}", path.display())),
)),
}
}
fn validate_replaceable_file(path: &Path, description: &str) -> Result<(), CliError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_file() => Ok(()),
Ok(_) => Err(CliError::Local(anyhow::anyhow!(
"{description} {} is not a regular file",
path.display()
))),
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(CliError::Local(
anyhow::Error::new(err).context(format!("inspect {description} {}", path.display())),
)),
}
}
fn commit_staged_file(staged: &Path, target: &Path, backup: &Path) -> io::Result<Option<PathBuf>> {
let old_file = match fs::symlink_metadata(target) {
Ok(_) => {
fs::rename(target, backup)?;
Some(backup.to_path_buf())
}
Err(err) if err.kind() == io::ErrorKind::NotFound => None,
Err(err) => return Err(err),
};
if let Err(err) = fs::rename(staged, target) {
if let Some(old_file) = old_file.as_deref() {
if let Err(restore_err) = fs::rename(old_file, target) {
return Err(io::Error::new(
restore_err.kind(),
format!(
"{err}; additionally failed to restore {}: {restore_err}",
target.display()
),
));
}
}
return Err(err);
}
Ok(old_file)
}
fn rollback_installed_files(installed: &[(PathBuf, Option<PathBuf>)]) -> io::Result<()> {
let mut rollback_error = None;
for (target, backup) in installed.iter().rev() {
match fs::remove_file(target) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
Err(err) => {
rollback_error.get_or_insert(err);
continue;
}
}
if let Some(backup) = backup {
if let Err(err) = fs::rename(backup, target) {
rollback_error.get_or_insert(err);
}
}
}
match rollback_error {
Some(err) => Err(err),
None => Ok(()),
}
}
fn is_generated_state_name(name: &str) -> bool {
name.strip_suffix(".txt").is_some_and(is_canonical_state_id)
}
fn restore_staged_stale_states(staged: &[(PathBuf, PathBuf)]) -> io::Result<()> {
let mut restore_error = None;
for (original, backup) in staged.iter().rev() {
if let Err(err) = fs::rename(backup, original) {
restore_error.get_or_insert(err);
}
}
match restore_error {
Some(err) => Err(err),
None => Ok(()),
}
}
/// Move stale generated states aside so replacement and cleanup form one
/// rollback boundary. User-owned files, including arbitrary non-`sN.txt`, stay.
fn stage_stale_states(
states_dir: &Path,
keep: &HashSet<String>,
transaction_id: uuid::Uuid,
) -> Result<Vec<(PathBuf, PathBuf)>, CliError> {
let entries = match fs::read_dir(states_dir) {
Ok(entries) => entries,
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(err) => {
return Err(CliError::Local(
anyhow::Error::new(err)
.context(format!("read states dir {}", states_dir.display())),
));
}
};
let mut staged = Vec::new();
for entry in entries {
let entry = match entry {
Ok(entry) => entry,
Err(err) => {
let restore_result = restore_staged_stale_states(&staged);
let message = match restore_result {
Ok(()) => format!("read stale state entry: {err}"),
Err(restore_err) => {
format!("read stale state entry: {err}; restore also failed: {restore_err}")
}
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
};
let name = entry.file_name();
let Some(name) = name.to_str() else { continue };
if !is_generated_state_name(name) || keep.contains(name) {
continue;
}
let original = entry.path();
let file_type = match entry.file_type() {
Ok(file_type) => file_type,
Err(err) => {
let restore_result = restore_staged_stale_states(&staged);
let message = match restore_result {
Ok(()) => format!("inspect stale state {}: {err}", original.display()),
Err(restore_err) => format!(
"inspect stale state {}: {err}; restore also failed: {restore_err}",
original.display()
),
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
};
if file_type.is_dir() {
let restore_result = restore_staged_stale_states(&staged);
let message = match restore_result {
Ok(()) => format!("stale state {} is a directory", original.display()),
Err(restore_err) => format!(
"stale state {} is a directory; restore also failed: {restore_err}",
original.display()
),
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
let backup = states_dir.join(format!(".{name}.{transaction_id}.stale"));
if let Err(err) = fs::rename(&original, &backup) {
let restore_result = restore_staged_stale_states(&staged);
let message = match restore_result {
Ok(()) => format!("stage stale state {}: {err}", original.display()),
Err(restore_err) => format!(
"stage stale state {}: {err}; restore also failed: {restore_err}",
original.display()
),
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
staged.push((original, backup));
}
Ok(staged)
}
fn commit_export_transaction(
staging_dir: &Path,
states_dir: &Path,
keep_states: &HashSet<String>,
transaction_id: uuid::Uuid,
replacements: Vec<(PathBuf, PathBuf, PathBuf)>,
) -> Result<(), CliError> {
let mut installed = Vec::with_capacity(replacements.len());
for (staged, target, backup) in replacements {
match commit_staged_file(&staged, &target, &backup) {
Ok(old_file) => installed.push((target, old_file)),
Err(err) => {
let rollback_result = rollback_installed_files(&installed);
let _ = fs::remove_dir_all(staging_dir);
let message = match rollback_result {
Ok(()) => format!("commit replacement {}: {err}", target.display()),
Err(rollback_err) => format!(
"commit replacement {}: {err}; rollback also failed: {rollback_err}",
target.display()
),
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
}
}
let stale_states = match stage_stale_states(states_dir, keep_states, transaction_id) {
Ok(stale) => stale,
Err(err) => {
let rollback_result = rollback_installed_files(&installed);
let _ = fs::remove_dir_all(staging_dir);
let message = match rollback_result {
Ok(()) => err.to_string(),
Err(rollback_err) => {
format!("{err}; replacement rollback also failed: {rollback_err}")
}
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
};
for (_, backup) in &installed {
if let Some(backup) = backup {
let _ = fs::remove_file(backup);
}
}
for (_, backup) in stale_states {
let _ = fs::remove_file(backup);
}
let _ = fs::remove_dir_all(staging_dir);
Ok(())
}
#[derive(Serialize)]
struct BundleState<'a> {
id: &'a str,
url: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
title: Option<&'a str>,
page: String,
#[serde(skip_serializing_if = "std::ops::Not::not")]
truncated: bool,
}
#[derive(Serialize)]
struct BundleTrace<'a> {
version: u32,
#[serde(skip_serializing_if = "Option::is_none")]
purpose: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
started_at: Option<&'a str>,
recorded_at: &'a str,
stopped_by: &'a bsk_protocol::tools::StopReason,
entry: &'a bsk_protocol::tools::TraceEntry,
recorder: &'a bsk_protocol::tools::RecorderInfo,
states: Vec<BundleState<'a>>,
steps: &'a [bsk_protocol::tools::StepV3],
}
pub(super) fn write_trace_bundle(output_dir: &Path, trace: &TraceV3) -> Result<PathBuf, CliError> {
let states_dir = states_dir_for_output(output_dir);
let trace_path = trace_json_path(output_dir);
if trace.version != TRACE_VERSION_V3 {
return Err(CliError::Local(anyhow::anyhow!(
"trace version {} is not supported (expected {})",
trace.version,
TRACE_VERSION_V3
)));
}
let mut state_ids = HashSet::with_capacity(trace.states.len());
for state in &trace.states {
if !is_canonical_state_id(&state.id) {
return Err(CliError::Local(anyhow::anyhow!(
"state id {:?} is not canonical (expected s1, s2, ...)",
state.id
)));
}
if !state_ids.insert(state.id.as_str()) {
return Err(CliError::Local(anyhow::anyhow!(
"duplicate state id {:?}",
state.id
)));
}
}
let mut disk_states = Vec::with_capacity(trace.states.len());
let mut state_writes = Vec::new();
let mut keep = HashSet::new();
for state in &trace.states {
let filename = format!("{}.txt", state.id);
let state_path = states_dir.join(&filename);
if state_path.parent() != Some(states_dir.as_path()) {
return Err(CliError::Local(anyhow::anyhow!(
"state path {} escapes states directory {}",
state_path.display(),
states_dir.display()
)));
}
keep.insert(filename.clone());
state_writes.push((filename.clone(), state.body.as_str()));
disk_states.push(BundleState {
id: &state.id,
url: &state.url,
title: state.title.as_deref(),
page: filename,
truncated: state.truncated,
});
}
let disk_trace = BundleTrace {
version: trace.version,
purpose: trace.purpose.as_deref(),
started_at: trace.started_at.as_deref(),
recorded_at: &trace.recorded_at,
stopped_by: &trace.stopped_by,
entry: &trace.entry,
recorder: &trace.recorder,
states: disk_states,
steps: &trace.steps,
};
let json = serde_json::to_string_pretty(&disk_trace)
.context("serialize trace JSON")
.map_err(CliError::Local)?;
validate_bundle_directory(output_dir, "output bundle directory")?;
fs::create_dir_all(output_dir)
.with_context(|| format!("create output dir {}", output_dir.display()))
.map_err(CliError::Local)?;
validate_bundle_directory(&states_dir, "states directory")?;
let _lock = acquire_export_lock(output_dir)?;
fs::create_dir_all(&states_dir)
.with_context(|| format!("create states dir {}", states_dir.display()))
.map_err(CliError::Local)?;
for (filename, _) in &state_writes {
validate_replaceable_file(&states_dir.join(filename), "state observation")?;
}
validate_replaceable_file(&trace_path, "trace JSON")?;
let transaction_id = uuid::Uuid::new_v4();
let staging_dir = output_dir.join(format!(".bsk-record-stage-{transaction_id}"));
let staging_states = staging_dir.join("states");
fs::create_dir_all(&staging_states)
.with_context(|| format!("create staging dir {}", staging_states.display()))
.map_err(CliError::Local)?;
let stage_result: Result<(), CliError> = (|| {
for (filename, body) in &state_writes {
let staged_path = staging_states.join(filename);
fs::write(&staged_path, body)
.with_context(|| {
format!("write staged state observation {}", staged_path.display())
})
.map_err(CliError::Local)?;
}
let staged_trace = staging_dir.join("trace.json");
fs::write(&staged_trace, format!("{json}\n"))
.with_context(|| format!("write staged trace to {}", staged_trace.display()))
.map_err(CliError::Local)?;
Ok(())
})();
if let Err(err) = stage_result {
let _ = fs::remove_dir_all(&staging_dir);
return Err(err);
}
let mut replacements: Vec<(PathBuf, PathBuf, PathBuf)> =
Vec::with_capacity(state_writes.len() + 1);
for (filename, _) in &state_writes {
replacements.push((
staging_states.join(filename),
states_dir.join(filename),
states_dir.join(format!(".{filename}.{transaction_id}.bak")),
));
}
replacements.push((
staging_dir.join("trace.json"),
trace_path.clone(),
output_dir.join(format!(".trace.json.{transaction_id}.bak")),
));
commit_export_transaction(
&staging_dir,
&states_dir,
&keep,
transaction_id,
replacements,
)?;
Ok(states_dir)
}
#[cfg(test)]
mod tests {
use std::fs;
use super::*;
use bsk_protocol::tools::{
NavigationCause, PageRefV2 as PageRef, RecorderInfo, StepCommonV3 as StepCommon,
StepResultV3 as StepResult, StepV3 as Step, StopReason, TRACE_VERSION_V2, TRACE_VERSION_V3,
TraceEntry, TraceStateV3, TraceV2, TraceV3, VOM_FORMAT_VERSION,
};
fn sample_trace(state_id: &str, body: &str) -> TraceV3 {
TraceV3 {
version: TRACE_VERSION_V3,
recorded_at: "2026-08-10T02:12:55.360Z".into(),
stopped_by: StopReason::UserFinish,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: state_id.into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: body.into(),
truncated: false,
}],
steps: vec![Step::Navigate {
common: StepCommon {
id: 1,
state: state_id.into(),
result: StepResult {
state: state_id.into(),
},
},
to: "https://example.com/".into(),
cause: NavigationCause::UserTyped,
}],
purpose: None,
started_at: None,
}
}
#[test]
fn write_trace_bundle_writes_dir_layout() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let trace = sample_trace("s1", "@vom 1\nL1 page");
let states_dir = write_trace_bundle(&output, &trace).unwrap();
let trace_path = output.join("trace.json");
assert!(trace_path.exists());
assert_eq!(states_dir, output.join("states"));
assert!(states_dir.join("s1.txt").exists());
let disk: serde_json::Value =
serde_json::from_str(&fs::read_to_string(&trace_path).unwrap()).unwrap();
assert_eq!(disk["states"][0]["page"], "s1.txt");
assert!(disk["states"][0].get("body").is_none());
assert_eq!(
fs::read_to_string(states_dir.join("s1.txt")).unwrap(),
"@vom 1\nL1 page"
);
// Re-exporting drops states from the previous run but keeps the
// directory in place for the concurrent writer.
fs::write(states_dir.join("s9.txt"), "stale").unwrap();
fs::write(states_dir.join("notes.md"), "keep me").unwrap();
fs::write(states_dir.join("reference.txt"), "user file").unwrap();
write_trace_bundle(&output, &trace).unwrap();
assert!(!states_dir.join("s9.txt").exists());
assert!(states_dir.join("s1.txt").exists());
assert!(states_dir.join("notes.md").exists());
assert_eq!(
fs::read_to_string(states_dir.join("reference.txt")).unwrap(),
"user file"
);
}
#[test]
fn write_trace_bundle_rejects_traversal_state_id() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let trace = sample_trace("../escaped", "@vom 1\nL1 page");
let err = write_trace_bundle(&output, &trace).unwrap_err();
assert!(err.to_string().contains("state id"));
assert!(!output.join("escaped.txt").exists());
assert!(!output.join("trace.json").exists());
}
#[test]
fn write_trace_bundle_rejects_duplicate_state_id() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let mut trace = sample_trace("s1", "@vom 1\nL1 first");
trace.states.push(TraceStateV3 {
body: "@vom 1\nL1 duplicate".into(),
..trace.states[0].clone()
});
let err = write_trace_bundle(&output, &trace).unwrap_err();
assert!(err.to_string().contains("duplicate state id"));
assert!(!output.exists());
}
#[test]
fn unsupported_version_preserves_existing_bundle() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
let mut replacement = sample_trace("s2", "replacement");
replacement.version += 1;
let err = write_trace_bundle(&output, &replacement).unwrap_err();
assert!(err.to_string().contains("not supported"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("states/s1.txt")).unwrap(),
"original"
);
assert!(!output.join("states/s2.txt").exists());
}
#[test]
fn state_replacement_failure_preserves_existing_bundle() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
fs::write(output.join("states/notes.md"), "keep me").unwrap();
fs::create_dir(output.join("states/s2.txt")).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s2", "replacement")).unwrap_err();
assert!(err.to_string().contains("state"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("states/s1.txt")).unwrap(),
"original"
);
assert_eq!(
fs::read_to_string(output.join("states/notes.md")).unwrap(),
"keep me"
);
}
#[test]
fn stale_state_cleanup_failure_rolls_back_replacement() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
fs::create_dir(output.join("states/s9.txt")).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s2", "replacement")).unwrap_err();
assert!(err.to_string().contains("stale state"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("states/s1.txt")).unwrap(),
"original"
);
assert!(!output.join("states/s2.txt").exists());
assert!(output.join("states/s9.txt").is_dir());
}
#[cfg(unix)]
#[test]
fn write_trace_bundle_rejects_symlinked_states_directory() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let external_states = dir.path().join("external-states");
fs::create_dir_all(&output).unwrap();
fs::create_dir_all(&external_states).unwrap();
fs::write(external_states.join("s9.txt"), "external").unwrap();
symlink(&external_states, output.join("states")).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s1", "replacement")).unwrap_err();
assert!(err.to_string().contains("symlink"));
assert_eq!(
fs::read_to_string(external_states.join("s9.txt")).unwrap(),
"external"
);
assert!(!external_states.join("s1.txt").exists());
assert!(!output.join("trace.json").exists());
}
#[cfg(unix)]
#[test]
fn write_trace_bundle_rejects_symlinked_output_directory() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let external_output = dir.path().join("external-output");
let output = dir.path().join("flow");
fs::create_dir_all(&external_output).unwrap();
fs::write(external_output.join("notes.md"), "external").unwrap();
symlink(&external_output, &output).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s1", "replacement")).unwrap_err();
assert!(err.to_string().contains("symlink"));
assert_eq!(
fs::read_to_string(external_output.join("notes.md")).unwrap(),
"external"
);
assert!(!external_output.join("trace.json").exists());
assert!(!external_output.join("states").exists());
}
#[test]
fn write_trace_v2_writes_single_json_and_prunes_stale_states() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
fs::create_dir_all(output.join("states")).unwrap();
fs::write(output.join("states/s1.txt"), "stale").unwrap();
let trace = TraceV2 {
recorded_at: "2026-08-10T02:12:55.360Z".into(),
started_at: None,
purpose: None,
entry: bsk_protocol::tools::TraceEntry {
start_url: "https://example.com/".into(),
},
pages: vec![PageRef {
id: "p1".into(),
url: "https://example.com/".into(),
title: None,
}],
steps: vec![],
};
write_trace_v2(&output, &trace).unwrap();
assert!(output.join("trace.json").exists());
assert!(!output.join("states/s1.txt").exists());
}
#[test]
fn write_trace_v2_stale_cleanup_failure_rolls_back_trace() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
fs::create_dir(output.join("states/s9.txt")).unwrap();
let trace = TraceV2 {
recorded_at: "2026-08-10T02:12:55.360Z".into(),
started_at: None,
purpose: None,
entry: bsk_protocol::tools::TraceEntry {
start_url: "https://example.com/".into(),
},
pages: vec![],
steps: vec![],
};
let err = write_trace_v2(&output, &trace).unwrap_err();
assert!(err.to_string().contains("stale state"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("states/s1.txt")).unwrap(),
"original"
);
assert!(output.join("states/s9.txt").is_dir());
}
#[test]
fn export_recorded_trace_accepts_v2() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let trace = RecordedTrace::V2(TraceV2 {
recorded_at: "2026-08-10T02:12:55.360Z".into(),
started_at: None,
purpose: None,
entry: bsk_protocol::tools::TraceEntry {
start_url: "https://example.com/".into(),
},
pages: vec![],
steps: vec![],
});
let exported = export_recorded_trace(&output, &trace).unwrap();
assert_eq!(exported.trace_version, TRACE_VERSION_V2);
assert!(exported.states_dir.is_none());
assert!(output.join("trace.json").exists());
}
#[test]
fn existing_trace_json_file_is_rejected_with_bundle_usage() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("trace.json");
fs::write(&output, "{}\n").unwrap();
let err = validate_record_output(&output).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("JSON file") || msg.contains("json file"),
"should name the legacy JSON file usage: {msg}"
);
assert!(
msg.contains("states/") || msg.contains("--output trace"),
"should hint at the new bundle directory usage: {msg}"
);
assert_eq!(fs::read_to_string(&output).unwrap(), "{}\n");
}
#[test]
fn missing_trace_json_path_is_rejected_as_legacy_output() {
let err = validate_record_output(Path::new("trace.json")).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("JSON file") || msg.contains("json file"),
"{msg}"
);
assert!(
msg.contains("states/") || msg.contains("--output trace"),
"{msg}"
);
}
#[test]
fn export_failure_keeps_recovery_for_retry() {
let _lock = crate::cli::record_recovery::test_env_lock();
let tmp = tempfile::tempdir().unwrap();
unsafe {
std::env::set_var(crate::daemon::paths::BSK_HOME_ENV, tmp.path());
}
let dir = tempfile::tempdir().unwrap();
let bad_output = dir.path().join("trace.json");
fs::write(&bad_output, "{}\n").unwrap();
let trace = RecordedTrace::V3(sample_trace("s1", "@vom 1\nL1 page"));
let err = export_with_recovery(&bad_output, &trace).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("recover") || msg.contains("record stop"),
"export failure should tell the user how to recover: {msg}"
);
assert!(
crate::cli::record_recovery::exists(),
"recovery data must survive a failed bundle export"
);
let recovered = crate::cli::record_recovery::load()
.unwrap()
.expect("recovery file");
assert_eq!(recovered, trace);
assert_eq!(fs::read_to_string(&bad_output).unwrap(), "{}\n");
let good_output = dir.path().join("bundle");
export_with_recovery(&good_output, &recovered).unwrap();
assert!(good_output.join("trace.json").exists());
assert!(good_output.join("states/s1.txt").exists());
assert!(
!crate::cli::record_recovery::exists(),
"successful export must delete recovery data"
);
unsafe {
std::env::remove_var(crate::daemon::paths::BSK_HOME_ENV);
}
}
}
+112
View File
@@ -0,0 +1,112 @@
//! Persist a completed Trace so a failed bundle export can be retried.
use std::fs;
use anyhow::{Context, Result};
use bsk_protocol::tools::RecordedTrace;
use crate::daemon::paths;
pub fn save(trace: &RecordedTrace) -> Result<()> {
let path = paths::record_recovery_path()?;
paths::ensure_bsk_home()?;
let json = serde_json::to_string_pretty(trace).context("serialize record recovery")?;
let tmp = path.with_extension("tmp");
fs::write(&tmp, format!("{json}\n")).with_context(|| format!("write {}", tmp.display()))?;
fs::rename(&tmp, &path).with_context(|| format!("commit {}", path.display()))?;
Ok(())
}
pub fn load() -> Result<Option<RecordedTrace>> {
let path = paths::record_recovery_path()?;
if !path.exists() {
return Ok(None);
}
let raw = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
let trace = serde_json::from_str(&raw).with_context(|| format!("parse {}", path.display()))?;
Ok(Some(trace))
}
pub fn exists() -> bool {
paths::record_recovery_path()
.map(|path| path.exists())
.unwrap_or(false)
}
pub fn clear() {
if let Ok(path) = paths::record_recovery_path() {
let _ = fs::remove_file(path);
}
}
#[cfg(test)]
pub(crate) fn test_env_lock() -> std::sync::MutexGuard<'static, ()> {
crate::daemon::paths::test_env_lock()
}
#[cfg(test)]
mod tests {
use super::*;
use bsk_protocol::tools::{
RecorderInfo, StopReason, TRACE_VERSION_V3, TraceEntry, TraceStateV3, TraceV3,
VOM_FORMAT_VERSION,
};
fn sample_trace() -> RecordedTrace {
RecordedTrace::V3(TraceV3 {
version: TRACE_VERSION_V3,
recorded_at: "2026-08-10T02:12:55.360Z".into(),
stopped_by: StopReason::UserFinish,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: "@vom 1\nL1 page".into(),
truncated: false,
}],
steps: vec![],
purpose: None,
started_at: None,
})
}
fn with_temp_home<F: FnOnce()>(f: F) {
let _lock = test_env_lock();
let tmp = tempfile::tempdir().unwrap();
unsafe {
std::env::set_var(crate::daemon::paths::BSK_HOME_ENV, tmp.path());
}
f();
clear();
unsafe {
std::env::remove_var(crate::daemon::paths::BSK_HOME_ENV);
}
}
#[test]
fn save_load_round_trips() {
with_temp_home(|| {
let trace = sample_trace();
save(&trace).unwrap();
assert!(exists());
assert_eq!(load().unwrap(), Some(trace));
});
}
#[test]
fn clear_removes_recovery_file() {
with_temp_home(|| {
save(&sample_trace()).unwrap();
clear();
assert!(!exists());
assert_eq!(load().unwrap(), None);
});
}
}
+1 -7
View File
@@ -59,15 +59,9 @@ fn started_at_unix_ms() -> String {
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
fn env_guard() -> &'static Mutex<()> {
static GUARD: Mutex<()> = Mutex::new(());
&GUARD
}
fn with_temp_home<F: FnOnce()>(f: F) {
let _lock = env_guard().lock().unwrap_or_else(|e| e.into_inner());
let _lock = crate::daemon::paths::test_env_lock();
let tmp = tempfile::tempdir().unwrap();
unsafe {
std::env::set_var(crate::daemon::paths::BSK_HOME_ENV, tmp.path());
+27
View File
@@ -50,6 +50,7 @@ pub mod reason {
pub const BORROW_CONFLICT: &str = "borrow_conflict";
pub const SCREENSHOT_CAPTURE_FAILED: &str = "screenshot_capture_failed";
pub const SESSION_BUSY: &str = crate::rpc_reason::SESSION_BUSY;
pub const RECORD_START_PAGE_UNREACHABLE: &str = "record_start_page_unreachable";
}
/// Per-code rendering metadata. The `summary` is what the CLI prints
@@ -269,6 +270,16 @@ pub fn info_for_error(code: ErrorCode, data: Option<&serde_json::Value>) -> Rend
),
exit_code: base.exit_code,
},
(
ErrorCode::CdpFailed | ErrorCode::Timeout | ErrorCode::InvalidParams,
reason::RECORD_START_PAGE_UNREACHABLE,
) => RenderInfo {
summary: "default start page https://example.com/ could not be opened",
hint: Some(
"that site may be unreachable from your network. Pass --url with a page you can open, e.g. `bsk record start --url https://<site>/`",
),
exit_code: base.exit_code,
},
_ => base,
}
}
@@ -378,6 +389,22 @@ mod tests {
assert_eq!(info.exit_code, 4);
}
#[test]
fn record_start_page_unreachable_overrides_cdp_failed_copy() {
let data = serde_json::json!({ "reason": reason::RECORD_START_PAGE_UNREACHABLE });
let info = info_for_error(ErrorCode::CdpFailed, Some(&data));
assert_eq!(
info.summary,
"default start page https://example.com/ could not be opened"
);
assert!(
info.hint.unwrap().contains("--url"),
"expected --url hint, got {:?}",
info.hint
);
assert_eq!(info.exit_code, 3);
}
#[test]
fn screenshot_capture_failed_overrides_cdp_failed_copy() {
let data = serde_json::json!({ "reason": reason::SCREENSHOT_CAPTURE_FAILED });
+21 -9
View File
@@ -100,6 +100,14 @@ pub fn record_session_path() -> Result<PathBuf> {
Ok(bsk_home()?.join("record-session.json"))
}
/// Path to a completed Trace saved before bundle export (`record-recovery.json`).
///
/// Kept until export succeeds so a failed `--output` write cannot drop the
/// recording the extension already returned.
pub fn record_recovery_path() -> Result<PathBuf> {
Ok(bsk_home()?.join("record-recovery.json"))
}
/// Windows named-pipe name. Include the resolved `BSK_HOME` path in the
/// token so test homes and custom installs do not share a predictable
/// per-username pipe.
@@ -119,6 +127,13 @@ pub fn pipe_name() -> String {
render_pipe_name(&user, &home)
}
/// Process-wide lock for tests that mutate `BSK_HOME`.
#[cfg(test)]
pub(crate) fn test_env_lock() -> std::sync::MutexGuard<'static, ()> {
static GUARD: std::sync::Mutex<()> = std::sync::Mutex::new(());
GUARD.lock().unwrap_or_else(|e| e.into_inner())
}
/// Render the Windows named-pipe name for a user/home pair. Kept
/// platform-agnostic so unit tests on any host can pin the output
/// character set.
@@ -136,19 +151,12 @@ fn render_pipe_name(user: &str, home: &str) -> String {
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use tempfile::TempDir;
/// Serialise tests that mutate the global `BSK_HOME` env var.
fn env_guard() -> &'static Mutex<()> {
static GUARD: Mutex<()> = Mutex::new(());
&GUARD
}
fn with_temp_home<F: FnOnce(&Path)>(f: F) {
let _lock = env_guard().lock().unwrap_or_else(|e| e.into_inner());
let _lock = test_env_lock();
let tmp = TempDir::new().unwrap();
// SAFETY: serialised by env_guard above.
// SAFETY: serialised by test_env_lock above.
unsafe {
std::env::set_var(BSK_HOME_ENV, tmp.path().join("bsk"));
}
@@ -187,6 +195,10 @@ mod tests {
record_session_path().unwrap(),
home.join("record-session.json")
);
assert_eq!(
record_recovery_path().unwrap(),
home.join("record-recovery.json")
);
});
}
+2 -2
View File
@@ -26,7 +26,7 @@ use crate::daemon::{
browsers::{BROWSER_LIVENESS_TICK, BROWSER_LIVENESS_TIMEOUT, EXTENSION_CONNECT_WAIT},
info as daemon_info, ipc, lockfile, paths,
sessions::{StopSessionError, forget_session, stop_session},
state::DaemonState,
state::{DaemonState, PROTOCOL_VERSION},
ws,
};
@@ -297,7 +297,7 @@ pub fn run_foreground(cfg: DaemonConfig) -> Result<()> {
ws_port,
sock_path: sock_path.clone(),
daemon_version: env!("CARGO_PKG_VERSION"),
protocol_version: "1.0",
protocol_version: PROTOCOL_VERSION,
};
let handler = ipc::full_handler(status, Arc::clone(&state));
+1 -1
View File
@@ -16,7 +16,7 @@ use super::start::DaemonConfig;
use super::ws::WsHandle;
pub const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
pub const PROTOCOL_VERSION: &str = "1.0";
pub const PROTOCOL_VERSION: &str = "1.1";
/// Lowest **protocol** version peers must speak (e.g. `"1.0"`).
pub const MIN_COMPATIBLE_PROTOCOL: &str = "1.0";
/// Legacy app-semver floor used only when `HandshakeResult.min_compatible_peer`
+10 -10
View File
@@ -108,12 +108,12 @@ async fn send_handshake_with_floors(
async fn handshake_ok_when_protocol_matches() {
let (handle, _sock) = spawn_daemon().await;
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake(&mut ws, "1.0", env!("CARGO_PKG_VERSION")).await;
let resp = send_handshake(&mut ws, "1.1", env!("CARGO_PKG_VERSION")).await;
let result: HandshakeResult = match resp.body {
ResponseBody::Ok(v) => serde_json::from_value(v).unwrap(),
ResponseBody::Err(e) => panic!("expected ok handshake, got {e:?}"),
};
assert_eq!(result.protocol_version, "1.0");
assert_eq!(result.protocol_version, "1.1");
assert_eq!(
result
.min_compatible_peer
@@ -135,7 +135,7 @@ async fn handshake_ok_when_app_versions_differ_but_protocol_matches() {
let (handle, _sock) = spawn_daemon().await;
let mut ws = open_ws(handle.ws_addr()).await;
let resp =
send_handshake_with_floors(&mut ws, "1.0", "9.9.9", Some("0.0.0"), Some("1.0")).await;
send_handshake_with_floors(&mut ws, "1.1", "9.9.9", Some("0.0.0"), Some("1.1")).await;
match resp.body {
ResponseBody::Ok(_) => {}
other => panic!("expected ok when protocol matches, got {other:?}"),
@@ -149,10 +149,10 @@ async fn handshake_skew_when_protocol_minor_differs() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.1",
"1.2",
env!("CARGO_PKG_VERSION"),
Some("0.0.0"),
Some("1.0"),
Some("1.1"),
)
.await;
match resp.body {
@@ -210,7 +210,7 @@ async fn handshake_legacy_ext_without_protocol_floor_still_ok() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.0",
"1.1",
env!("CARGO_PKG_VERSION"),
Some("0.1.0"),
None,
@@ -235,7 +235,7 @@ async fn status_surfaces_version_skew_for_skewed_browser() {
browser_name: "chrome".into(),
browser_version: "131.0".into(),
extension_version: "9.9.9".into(),
extension_protocol_version: "1.1".into(),
extension_protocol_version: "1.2".into(),
label: "Older".into(),
sink: bsk::daemon::browsers::BrowserSink { tx },
pending: Mutex::new(bsk::daemon::browsers::Pending::default()),
@@ -263,8 +263,8 @@ async fn status_surfaces_version_skew_for_skewed_browser() {
.iter()
.find(|s| s.instance_id == "skew-only-test")
.expect("status must list our skew client");
assert_eq!(skew.client_protocol_version, "1.1");
assert_eq!(skew.server_protocol_version, "1.0");
assert_eq!(skew.client_protocol_version, "1.2");
assert_eq!(skew.server_protocol_version, "1.1");
assert_eq!(skew.client_version, "9.9.9");
let entry = status
.browsers
@@ -281,7 +281,7 @@ async fn handshake_rejects_when_local_below_peer_min_compatible_protocol() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.0",
"1.1",
env!("CARGO_PKG_VERSION"),
Some("0.0.0"),
Some("99.0.0"),
@@ -0,0 +1,327 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use bsk::daemon::info::DaemonInfo;
use bsk::daemon::paths::BSK_HOME_ENV;
use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::system::{HandshakeParams, HandshakeResult};
use bsk_protocol::tools::{
RecordStopResult, RecordedTrace, RecorderInfo, SessionStartParams, SessionStartResult,
StopReason, TraceEntry, TraceStateV3, TraceV3, VOM_FORMAT_VERSION,
};
use bsk_protocol::{BrowserPeerInfo, Frame, Method, RequestFrame, ResponseBody, ResponseFrame};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::process::Command;
use tokio_tungstenite::tungstenite::handshake::client::generate_key;
use tokio_tungstenite::tungstenite::http::Request;
use tokio_tungstenite::tungstenite::protocol::Message;
const TEST_EXT_ID: &str = "abcdefghijklmnopabcdefghijklmnop";
type Ws =
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
fn bsk_bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_bsk"))
}
async fn connect_extension(addr: std::net::SocketAddr) -> Ws {
let origin = format!("chrome-extension://{TEST_EXT_ID}");
let request = Request::builder()
.method("GET")
.uri(format!("ws://{addr}/"))
.header("Host", addr.to_string())
.header("Upgrade", "websocket")
.header("Connection", "Upgrade")
.header("Sec-WebSocket-Version", "13")
.header("Sec-WebSocket-Key", generate_key())
.header("Origin", origin)
.body(())
.unwrap();
tokio_tungstenite::connect_async(request).await.unwrap().0
}
async fn handshake(ws: &mut Ws) {
let request = RequestFrame {
id: "handshake".into(),
method: Method::SystemHandshake,
params: Some(
serde_json::to_value(HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.10".parse().unwrap(),
protocol_version: "1.1".into(),
instance_id: TEST_EXT_ID.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
version: "131.0".into(),
},
min_compatible_peer: Some("0.1.10".parse().unwrap()),
min_compatible_protocol: Some("1.1".into()),
label: "Record export recovery test".into(),
})
.unwrap(),
),
};
ws.send(Message::Text(serde_json::to_string(&request).unwrap()))
.await
.unwrap();
let Message::Text(response) = ws.next().await.unwrap().unwrap() else {
panic!("expected text handshake response");
};
let response: ResponseFrame = serde_json::from_str(&response).unwrap();
let ResponseBody::Ok(value) = response.body else {
panic!("handshake failed");
};
let _: HandshakeResult = serde_json::from_value(value).unwrap();
}
fn sample_trace() -> TraceV3 {
TraceV3 {
version: 3,
purpose: Some("export recovery".into()),
started_at: Some("2026-08-12T10:00:00Z".into()),
recorded_at: "2026-08-12T10:01:00Z".into(),
stopped_by: StopReason::CliStop,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: "@vom 1\nL1 page".into(),
truncated: false,
}],
steps: vec![],
}
}
fn run_extension(
mut ws: Ws,
record_stop_calls: Arc<AtomicUsize>,
session_stop_calls: Arc<AtomicUsize>,
) {
tokio::spawn(async move {
while let Some(Ok(Message::Text(text))) = ws.next().await {
let Ok(Frame::Request(request)) = serde_json::from_str::<Frame>(&text) else {
continue;
};
let body = match request.method {
Method::ToolSessionStart => {
let _: SessionStartParams =
serde_json::from_value(request.params.clone().unwrap()).unwrap();
ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(100),
})
.unwrap(),
)
}
Method::ToolRecordStop => {
record_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(
serde_json::to_value(RecordStopResult {
trace: RecordedTrace::V3(sample_trace()),
})
.unwrap(),
)
}
Method::ToolSessionStop => {
session_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(json!({}))
}
_ => ResponseBody::Err(bsk_protocol::RpcError {
code: bsk_protocol::ErrorCode::ProtocolError,
message: format!("unexpected method {:?}", request.method),
data: None,
}),
};
let response = ResponseFrame {
id: request.id,
body,
};
ws.send(Message::Text(serde_json::to_string(&response).unwrap()))
.await
.unwrap();
}
});
}
async fn start_session(sock: &Path) -> String {
#[derive(serde::Deserialize)]
struct StartResult {
session_id: String,
}
let mut client = IpcClient::connect(sock).await.unwrap();
client
.call::<(), StartResult>(
"session-start",
Method::SessionStart,
None,
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap()
.session_id
}
fn write_record_session(home: &Path, session_id: &str) {
std::fs::create_dir_all(home).unwrap();
std::fs::write(
home.join("record-session.json"),
format!("{{\"session_id\":\"{session_id}\",\"started_at\":\"1\"}}\n"),
)
.unwrap();
}
async fn run_record_stop(home: &Path, output: &Path) -> std::process::Output {
Command::new(bsk_bin())
.arg("record")
.arg("stop")
.arg("--output")
.arg(output)
.env(BSK_HOME_ENV, home)
.output()
.await
.unwrap()
}
async fn setup_daemon(temp: &Path) -> (bsk::daemon::DaemonHandle, PathBuf, PathBuf) {
let home = temp.join("bsk-home");
let sock = temp.join("daemon.sock");
let daemon = daemon::run(DaemonConfig::new(0), Some(sock.clone()))
.await
.unwrap();
bsk::daemon::info::write_to_path(
&DaemonInfo::now(
std::process::id(),
sock.clone(),
daemon.ws_addr().port(),
env!("CARGO_PKG_VERSION"),
),
&home.join("daemon.json"),
)
.unwrap();
(daemon, home, sock)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn existing_trace_json_file_is_rejected_before_record_stop() {
let temp = tempfile::Builder::new()
.prefix("bsk-record-json-")
.tempdir()
.unwrap();
let (daemon, home, sock) = setup_daemon(temp.path()).await;
let output = temp.path().join("trace.json");
std::fs::write(&output, "{}\n").unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
write_record_session(&home, &session_id);
let first = run_record_stop(&home, &output).await;
let stderr = String::from_utf8_lossy(&first.stderr);
assert!(
!first.status.success(),
"legacy JSON --output should fail: {stderr}"
);
assert!(
stderr.contains("JSON file") || stderr.contains("bundle directory"),
"{stderr}"
);
let state: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(home.join("record-session.json")).unwrap())
.unwrap();
assert_eq!(state["session_id"], session_id);
assert_eq!(
record_stop_calls.load(Ordering::SeqCst),
0,
"existing trace.json must be rejected before the extension returns the Trace"
);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 0);
assert_eq!(std::fs::read_to_string(&output).unwrap(), "{}\n");
daemon.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_bundle_export_can_be_recovered() {
let temp = tempfile::Builder::new()
.prefix("bsk-record-recov-")
.tempdir()
.unwrap();
let (daemon, home, sock) = setup_daemon(temp.path()).await;
let output = temp.path().join("trace");
std::fs::create_dir_all(&output).unwrap();
std::fs::write(output.join("states"), "not a directory\n").unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
write_record_session(&home, &session_id);
let first = run_record_stop(&home, &output).await;
let stderr = String::from_utf8_lossy(&first.stderr);
assert!(
!first.status.success(),
"blocked states path should fail export: {stderr}"
);
assert!(
home.join("record-recovery.json").exists(),
"completed Trace must be saved before the failed export"
);
assert!(
!home.join("record-session.json").exists(),
"session state is cleared after the extension returns the Trace"
);
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 1);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 1);
std::fs::remove_file(output.join("states")).unwrap();
let second = run_record_stop(&home, &output).await;
assert!(
second.status.success(),
"recovery export failed: {}",
String::from_utf8_lossy(&second.stderr)
);
assert_eq!(
record_stop_calls.load(Ordering::SeqCst),
1,
"recovery must not call the extension again"
);
assert!(output.join("trace.json").exists());
assert!(output.join("states/s1.txt").exists());
assert!(
!home.join("record-recovery.json").exists(),
"successful export must delete recovery data"
);
daemon.shutdown().await;
}
+271
View File
@@ -0,0 +1,271 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use bsk::cli::record_state;
use bsk::daemon::info::DaemonInfo;
use bsk::daemon::paths::BSK_HOME_ENV;
use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::system::{HandshakeParams, HandshakeResult};
use bsk_protocol::tools::{
RecordStopResult, RecordedTrace, RecorderInfo, SessionStartParams, SessionStartResult,
StopReason, TraceEntry, TraceStateV3, TraceV3, VOM_FORMAT_VERSION,
};
use bsk_protocol::{
BrowserPeerInfo, ErrorCode, Frame, Method, RequestFrame, ResponseBody, ResponseFrame, RpcError,
};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::process::Command;
use tokio_tungstenite::tungstenite::handshake::client::generate_key;
use tokio_tungstenite::tungstenite::http::Request;
use tokio_tungstenite::tungstenite::protocol::Message;
const TEST_EXT_ID: &str = "abcdefghijklmnopabcdefghijklmnop";
type Ws =
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
fn bsk_bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_bsk"))
}
async fn connect_extension(addr: std::net::SocketAddr) -> Ws {
let origin = format!("chrome-extension://{TEST_EXT_ID}");
let request = Request::builder()
.method("GET")
.uri(format!("ws://{addr}/"))
.header("Host", addr.to_string())
.header("Upgrade", "websocket")
.header("Connection", "Upgrade")
.header("Sec-WebSocket-Version", "13")
.header("Sec-WebSocket-Key", generate_key())
.header("Origin", origin)
.body(())
.unwrap();
tokio_tungstenite::connect_async(request).await.unwrap().0
}
async fn handshake(ws: &mut Ws) {
let request = RequestFrame {
id: "handshake".into(),
method: Method::SystemHandshake,
params: Some(
serde_json::to_value(HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.10".parse().unwrap(),
protocol_version: "1.1".into(),
instance_id: TEST_EXT_ID.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
version: "131.0".into(),
},
min_compatible_peer: Some("0.1.10".parse().unwrap()),
min_compatible_protocol: Some("1.1".into()),
label: "Record stop retry test".into(),
})
.unwrap(),
),
};
ws.send(Message::Text(serde_json::to_string(&request).unwrap()))
.await
.unwrap();
let Message::Text(response) = ws.next().await.unwrap().unwrap() else {
panic!("expected text handshake response");
};
let response: ResponseFrame = serde_json::from_str(&response).unwrap();
let ResponseBody::Ok(value) = response.body else {
panic!("handshake failed");
};
let _: HandshakeResult = serde_json::from_value(value).unwrap();
}
fn sample_trace() -> TraceV3 {
TraceV3 {
version: 3,
purpose: Some("retry record stop".into()),
started_at: Some("2026-08-12T10:00:00Z".into()),
recorded_at: "2026-08-12T10:01:00Z".into(),
stopped_by: StopReason::CliStop,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: "@vom 1\nL1 page".into(),
truncated: false,
}],
steps: vec![],
}
}
fn run_extension(
mut ws: Ws,
record_stop_calls: Arc<AtomicUsize>,
session_stop_calls: Arc<AtomicUsize>,
) {
tokio::spawn(async move {
while let Some(Ok(Message::Text(text))) = ws.next().await {
let Ok(Frame::Request(request)) = serde_json::from_str::<Frame>(&text) else {
continue;
};
let body = match request.method {
Method::ToolSessionStart => {
let _: SessionStartParams =
serde_json::from_value(request.params.clone().unwrap()).unwrap();
ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(100),
})
.unwrap(),
)
}
Method::ToolRecordStop => {
if record_stop_calls.fetch_add(1, Ordering::SeqCst) == 0 {
ResponseBody::Err(RpcError {
code: ErrorCode::ProtocolError,
message: "recording remains active; retry record stop".into(),
data: Some(json!({ "recording": true, "retryable": true })),
})
} else {
ResponseBody::Ok(
serde_json::to_value(RecordStopResult {
trace: RecordedTrace::V3(sample_trace()),
})
.unwrap(),
)
}
}
Method::ToolSessionStop => {
session_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(json!({}))
}
_ => ResponseBody::Err(RpcError {
code: ErrorCode::ProtocolError,
message: format!("unexpected method {:?}", request.method),
data: None,
}),
};
let response = ResponseFrame {
id: request.id,
body,
};
ws.send(Message::Text(serde_json::to_string(&response).unwrap()))
.await
.unwrap();
}
});
}
async fn start_session(sock: &Path) -> String {
#[derive(serde::Deserialize)]
struct StartResult {
session_id: String,
}
let mut client = IpcClient::connect(sock).await.unwrap();
client
.call::<(), StartResult>(
"session-start",
Method::SessionStart,
None,
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap()
.session_id
}
async fn run_record_stop(home: &Path, output: &Path) -> std::process::Output {
Command::new(bsk_bin())
.arg("record")
.arg("stop")
.arg("--output")
.arg(output)
.env(BSK_HOME_ENV, home)
.output()
.await
.unwrap()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_record_stop_keeps_state_and_session_for_retry() {
// Keep the Unix-domain socket below macOS's SUN_LEN even from a deep worktree.
let temp = tempfile::Builder::new()
.prefix("bsk-record-stop-")
.tempdir()
.unwrap();
let home = temp.path().join("bsk-home");
let sock = temp.path().join("daemon.sock");
let output = temp.path().join("trace");
let daemon = daemon::run(DaemonConfig::new(0), Some(sock.clone()))
.await
.unwrap();
bsk::daemon::info::write_to_path(
&DaemonInfo::now(
std::process::id(),
sock.clone(),
daemon.ws_addr().port(),
env!("CARGO_PKG_VERSION"),
),
&home.join("daemon.json"),
)
.unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
unsafe {
std::env::set_var(BSK_HOME_ENV, &home);
}
record_state::write(&session_id).unwrap();
let first = run_record_stop(&home, &output).await;
assert!(
!first.status.success(),
"first stop unexpectedly succeeded: {}",
String::from_utf8_lossy(&first.stdout)
);
assert_eq!(record_state::read().unwrap().session_id, session_id);
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 1);
assert_eq!(
session_stop_calls.load(Ordering::SeqCst),
0,
"failed record stop must leave the session active"
);
let second = run_record_stop(&home, &output).await;
assert!(
second.status.success(),
"retry failed: {}",
String::from_utf8_lossy(&second.stderr)
);
assert!(record_state::read().is_err());
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 2);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 1);
assert!(output.join("trace.json").exists());
assert!(output.join("states/s1.txt").exists());
unsafe {
std::env::remove_var(BSK_HOME_ENV);
}
daemon.shutdown().await;
}
+1 -1
View File
@@ -61,7 +61,7 @@ fn bsk_status_json_returns_structured_payload() {
assert!(parsed["pid"].as_u64().unwrap() > 0);
assert!(!parsed["daemon_version"].as_str().unwrap().is_empty());
assert_eq!(parsed["protocol_version"], "1.0");
assert_eq!(parsed["protocol_version"], "1.1");
assert!(parsed["sock_path"].as_str().is_some());
assert_eq!(parsed["browsers"], serde_json::json!([]));
assert_eq!(parsed["sessions"], serde_json::json!([]));
+2 -2
View File
@@ -53,7 +53,7 @@ pub async fn send_handshake(
let params = HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.0-dev.0".parse().unwrap(),
protocol_version: "1.0".into(),
protocol_version: "1.1".into(),
instance_id: instance_id.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
@@ -91,7 +91,7 @@ async fn ws_handshake_registers_browser_in_state() {
let mut ws = connect_ext(handle.ws_addr(), &origin).await;
let result = send_handshake(&mut ws, TEST_EXT_ID).await;
assert_eq!(result.server, "browser-skill-daemon");
assert_eq!(result.protocol_version, "1.0");
assert_eq!(result.protocol_version, "1.1");
let state = handle.state();
let browsers = state.browsers.snapshot();
@@ -6,12 +6,26 @@
"session_id"
],
"properties": {
"max_page_tokens": {
"type": [
"integer",
"null"
],
"format": "uint32",
"minimum": 0.0
},
"purpose": {
"type": [
"string",
"null"
]
},
"redact_values": {
"type": [
"boolean",
"null"
]
},
"session_id": {
"type": "string"
},
@@ -22,6 +36,15 @@
],
"format": "int64"
},
"trace_version": {
"description": "Desired trace export format. Omitted means v2; `3` requests v3.",
"type": [
"integer",
"null"
],
"format": "uint32",
"minimum": 0.0
},
"url": {
"type": [
"string",
+272
View File
@@ -32,6 +32,13 @@ pub struct RecordStartParams {
pub url: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub purpose: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_page_tokens: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub redact_values: Option<bool>,
/// Desired trace export format. Omitted means v2; `3` requests v3.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub trace_version: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
@@ -267,6 +274,232 @@ mod tests {
use super::*;
use serde_json::json;
fn sample_common(id: u32, state: &str, result_state: &str) -> StepCommonV3 {
StepCommonV3 {
id,
state: state.into(),
result: StepResultV3 {
state: result_state.into(),
},
}
}
fn sample_target() -> TargetDescriptorV3 {
TargetDescriptorV3 {
element_ref: Some("e21".into()),
role: Some("button".into()),
name: Some("发布".into()),
ctx: Some("金桔柠檬 6 号".into()),
unmatched: false,
}
}
fn sample_trace() -> TraceV3 {
TraceV3 {
version: TRACE_VERSION_V3,
purpose: Some("把草稿商品发布上架".into()),
started_at: Some("2026-08-10T02:10:41.080Z".into()),
recorded_at: "2026-08-10T02:12:55.360Z".into(),
stopped_by: StopReason::UserFinish,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![
TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example Domain".into()),
body: "@vom 1\nRootWebArea \"Example Domain\"".into(),
truncated: false,
},
TraceStateV3 {
id: "s2".into(),
url: "https://shop.example.com/products?status=draft".into(),
title: Some("商品管理".into()),
body: "@vom 1\nRootWebArea \"商品管理\"".into(),
truncated: false,
},
],
steps: vec![
StepV3::Navigate {
common: sample_common(1, "s1", "s2"),
to: "https://shop.example.com/products?status=draft".into(),
cause: NavigationCause::UserTyped,
},
StepV3::Fill {
common: sample_common(2, "s2", "s2"),
target: TargetDescriptorV3 {
element_ref: Some("e12".into()),
role: Some("textbox".into()),
name: Some("搜索商品".into()),
ctx: None,
unmatched: false,
},
value: "金桔柠檬".into(),
commit: FillCommit::Enter,
redacted: false,
},
StepV3::Click {
common: sample_common(3, "s2", "s2"),
target: sample_target(),
},
],
}
}
#[test]
fn step_click_round_trips() {
let step = StepV3::Click {
common: sample_common(1, "s1", "s2"),
target: sample_target(),
};
let v = serde_json::to_value(&step).unwrap();
assert_eq!(v.get("op").and_then(|v| v.as_str()), Some("click"));
assert_eq!(v.get("state").and_then(|v| v.as_str()), Some("s1"));
assert_eq!(v["result"]["state"], "s2");
assert_eq!(v["target"]["ref"], "e21");
let round: StepV3 = serde_json::from_value(v).unwrap();
assert_eq!(round, step);
}
#[test]
fn step_fill_with_commit_round_trips() {
let step = StepV3::Fill {
common: sample_common(2, "s2", "s3"),
target: TargetDescriptorV3 {
element_ref: Some("e12".into()),
role: Some("textbox".into()),
name: Some("搜索商品".into()),
ctx: None,
unmatched: false,
},
value: "browser skill".into(),
commit: FillCommit::Enter,
redacted: false,
};
let v = serde_json::to_value(&step).unwrap();
assert_eq!(v["op"], "fill");
assert_eq!(v["commit"], "enter");
assert!(v.get("redacted").is_none());
let round: StepV3 = serde_json::from_value(v).unwrap();
assert_eq!(round, step);
}
#[test]
fn step_fill_password_is_redacted() {
let step = StepV3::Fill {
common: sample_common(1, "s1", "s1"),
target: TargetDescriptorV3 {
element_ref: Some("e3".into()),
role: Some("textbox".into()),
name: Some("密码".into()),
ctx: None,
unmatched: false,
},
value: "***".into(),
commit: FillCommit::Blur,
redacted: true,
};
let v = serde_json::to_value(&step).unwrap();
assert_eq!(v["value"], "***");
assert_eq!(v["redacted"], true);
}
#[test]
fn step_navigate_with_cause_round_trips() {
let step = StepV3::Navigate {
common: sample_common(1, "s1", "s2"),
to: "https://example.com".into(),
cause: NavigationCause::UserTyped,
};
let v = serde_json::to_value(&step).unwrap();
assert_eq!(v["op"], "navigate");
assert_eq!(v["cause"], "user_typed");
let round: StepV3 = serde_json::from_value(v).unwrap();
assert_eq!(round, step);
}
#[test]
fn trace_v3_round_trips() {
let trace = sample_trace();
let v = serde_json::to_value(&trace).unwrap();
assert_eq!(v.get("version").and_then(|v| v.as_u64()), Some(3));
assert!(v.get("pages").is_none());
assert_eq!(v["recorder"]["vom"], 1);
assert_eq!(v["states"].as_array().unwrap().len(), 2);
let round: TraceV3 = serde_json::from_value(v).unwrap();
assert_eq!(round, trace);
}
#[test]
fn default_fields_are_omitted() {
let step = StepV3::Click {
common: sample_common(1, "s1", "s1"),
target: TargetDescriptorV3 {
element_ref: Some("e1".into()),
role: Some("button".into()),
name: Some("OK".into()),
ctx: None,
unmatched: false,
},
};
let v = serde_json::to_value(&step).unwrap();
assert!(v.get("unmatched").is_none());
assert!(v["target"].get("ctx").is_none());
assert!(v["target"].get("unmatched").is_none());
}
#[test]
fn record_start_trace_options_are_explicit_and_optional() {
let legacy = RecordStartParams {
session_id: "session".into(),
tab_id: None,
url: None,
purpose: None,
max_page_tokens: None,
redact_values: None,
trace_version: None,
};
let legacy_value = serde_json::to_value(legacy).unwrap();
assert!(legacy_value.get("trace_version").is_none());
assert!(legacy_value.get("max_page_tokens").is_none());
let v3 = RecordStartParams {
session_id: "session".into(),
tab_id: None,
url: None,
purpose: None,
max_page_tokens: Some(4_000),
redact_values: Some(true),
trace_version: Some(TRACE_VERSION_V3),
};
let v3_value = serde_json::to_value(v3).unwrap();
assert_eq!(v3_value["trace_version"], TRACE_VERSION_V3);
assert_eq!(v3_value["max_page_tokens"], 4_000);
assert_eq!(v3_value["redact_values"], true);
}
#[test]
fn unmatched_target_serializes_flag() {
let step = StepV3::Click {
common: sample_common(1, "s1", "s2"),
target: TargetDescriptorV3 {
element_ref: None,
role: Some("button".into()),
name: Some("发布".into()),
ctx: None,
unmatched: true,
},
};
let v = serde_json::to_value(&step).unwrap();
assert_eq!(v["target"]["unmatched"], true);
assert!(v["target"].get("ref").is_none());
}
#[test]
fn recorded_trace_classifies_v2_and_v3() {
let v2 = json!({
@@ -452,4 +685,43 @@ mod tests {
assert!(schema["definitions"].get("StepV2").is_some());
assert!(schema["definitions"].get("StepV3").is_some());
}
#[test]
fn extension_trace_deserializes() {
let v = json!({
"version": 3,
"recorded_at": "2026-07-21T08:00:00Z",
"started_at": "2026-07-21T07:59:00Z",
"purpose": "demo",
"stopped_by": "user_finish",
"entry": { "start_url": "https://example.com/editor" },
"recorder": { "bsk": "0.1.10", "vom": 1 },
"states": [
{ "id": "s1", "url": "https://example.com/editor", "body": "@vom 1" },
{ "id": "s2", "url": "https://example.com/p/99", "body": "@vom 1" }
],
"steps": [
{
"op": "fill",
"id": 1,
"state": "s1",
"result": { "state": "s1" },
"target": { "ref": "e1", "role": "textbox", "name": "标题" },
"value": "hello",
"commit": "blur"
},
{
"op": "click",
"id": 2,
"state": "s1",
"result": { "state": "s2" },
"target": { "ref": "e2", "role": "button", "name": "发布" }
}
]
});
let trace: TraceV3 = serde_json::from_value(v).unwrap();
assert_eq!(trace.version, 3);
assert_eq!(trace.states.len(), 2);
assert_eq!(trace.steps.len(), 2);
}
}
@@ -48,7 +48,7 @@
"record": {
"sectionTitle": "Action recording",
"cardDesc": "Record your actions for Agent reference",
"sectionHint": "Generates a recording instruction — send it to your Agent to start recording your actions; the Agent can then use the recording to automate similar tasks.",
"sectionHint": "Generates a recording instruction — send it to your Agent to start recording your actions; the Agent can then use the resulting trace bundle to automate similar tasks.",
"topicLabel": "Topic (optional)",
"topicPlaceholder": "e.g. Publish an article",
"startUrlLabel": "Starting URL (optional)",
@@ -56,7 +56,7 @@
"copyButton": "Copy recording instructions",
"copied": "Copied",
"hintDisconnected": "Available when connected",
"promptTemplate": "Use BrowserSkill's `bsk` CLI to record the browser actions I'm about to demonstrate.\n\nSteps:\n1. Run: {{command}}\n2. Once you confirm it's started, I'll operate in the opened window myself.\n3. After I click Finish, read the returned trace.json and summarize the actions (pages + steps); don't re-run or \"verify\"."
"promptTemplate": "Use BrowserSkill's `bsk` CLI to record the browser actions I'm about to demonstrate.\n\nSteps:\n1. Run: {{command}}\n2. Once you confirm it's started, I'll operate in the opened window myself.\n3. After I click Finish, read the written `./trace` bundle (`trace.json` + `states/`) and summarize what I did."
}
},
"controlOverlay": {
@@ -48,7 +48,7 @@
"record": {
"sectionTitle": "操作录制",
"cardDesc": "录制你的操作,供 Agent 参考",
"sectionHint": "生成一段录制指令,发给 Agent 后可以启动插件录制用户操作,之后 Agent 可以参考录制结果自动完成同类任务。",
"sectionHint": "生成一段录制指令,发给 Agent 后可以启动插件录制用户操作,之后 Agent 可以参考写出的 trace 目录包自动完成同类任务。",
"topicLabel": "操作主题(可选)",
"topicPlaceholder": "例如:发布一篇文章",
"startUrlLabel": "起始网址(可选)",
@@ -56,7 +56,7 @@
"copyButton": "复制录制指令",
"copied": "已复制",
"hintDisconnected": "连接后可用",
"promptTemplate": "用 BrowserSkill 的 `bsk` CLI 录制我接下来的浏览器操作。\n\n步骤:\n1. 执行:{{command}}\n2. 你确认已开始后,我会在打开的窗口里自己操作。\n3. 我点“结束”后,读取返回的 trace.json 并总结操作(pages + steps),不要重跑或“验证”。"
"promptTemplate": "用 BrowserSkill 的 `bsk` CLI 录制我接下来的浏览器操作。\n\n步骤:\n1. 执行:{{command}}\n2. 你确认已开始后,我会在打开的窗口里自己操作。\n3. 我点“结束”后,读取写出的 `./trace` 目录包(`trace.json` + `states/`),总结我做了什么。"
}
},
"controlOverlay": {
+14 -6
View File
@@ -265,19 +265,27 @@ retry — complete the task autonomously or stop gracefully.
### Recording — `bsk record`
Capture the user's own actions in the Agent Window to a `trace.json`, for later LLM-driven automation:
Capture the user's own actions in the Agent Window to a **trace bundle**, for later LLM-driven automation. New CLI builds request **trace v3** (page observations + action chain); older extensions may still return **trace v2** (actions only), which the CLI exports as a single `trace.json` without `states/`.
```bash
bsk record start --browser <instance-id-or-label> [--url https://…] [--purpose "publish a wiki doc"] [--output trace.json]
bsk record start --browser <instance-id-or-label> \
[--url https://…] [--purpose "publish a wiki doc"] \
[--max-page-tokens 3000] [--redact-values] \
[--output trace]
# `--url` is optional; default https://example.com/ when omitted (must be http(s)).
# Blocks until the user clicks Finish in the recording panel, then writes ./trace.json and closes the window.
# Blocks until the user clicks Finish in the recording panel, then writes:
# trace/trace.json — action chain (+ state index when v3)
# trace/states/ — v3 only: one `sN.txt` observe snapshot per settled page state
bsk record stop [--output trace.json] # terminal fallback if the browser panel is unavailable
bsk record stop [--output trace] # terminal fallback if the browser panel is unavailable
```
- The trace is a **record-only action log** (a `pages[]` dictionary + `navigate`/`click`/`fill`/`select`/`press` steps with `target` descriptors). It records *what the user did*; deciding which inputs are variable is left to the executing agent.
- **v3 bundle (preferred):** `--output` is a directory (default `./trace`) containing `trace.json` and `states/`. `trace.json` lists `states[]` (page observation ids) and `steps[]` (each step binds `state` = observe snapshot *before* the action and `result.state` = snapshot *after* settle). Page bodies live in `states/sN.txt` using the same VOM format as `bsk observe`.
- **v2 fallback:** when the connected extension is older, export may contain only legacy `trace.json` with `pages[]` action context and no `states/` observation files. Update the extension for full v3 bundles.
- Each `states[]` entry is one **settled page observation** (captured after recording start, navigation landing, or action settle — not on a timer).
- `target.ref` values like `@e12` exist **only inside** the bundle for disambiguation; do **not** copy `@eN` refs into SKILL.md or agent runbooks — use visible names from the observation text instead.
- `--purpose` is optional context metadata; it does **not** change what gets captured.
- There is **no** `bsk replay` — to redo a flow, read the trace and reuse the existing `session` / `snapshot` / `@eN` / `click` / `fill` tools. Follow **Stop when the goal is met**.
- `--redact-values` masks all form values in page files as `[filled]` / `[empty]`.
- Do **not** record on banking/SSO/password-manager pages; passwords are redacted but traces may still contain sensitive text.
## Error handling