Slice 5: record browser tasks with tyto learn

README.md

@@ -51,9 +51,9 @@ npm run test:live    # opt-in: needs agent-browser installed
 |---|---|
 | `@tyto/core` | Recipes, lint, verify, brief, find, redaction, ports, fakes (pure) |
 | `@tyto/llm` | OpenAI-compatible and Anthropic HTTP model adapters |
-| `@tyto/agent-browser` | Runs the agent-browser CLI (argv, batch JSON) |
-| `@tyto/store` | Recipes, session locks, replay config under `~/.tyto` |
-| `@tyto/cli` | The `tyto` command (`open`, `brief`, `find`, actions, `run`, `test`, `recipes`) |
+| `@tyto/agent-browser` | Runs the agent-browser CLI (argv, batch JSON); reads its event stream |
+| `@tyto/store` | Recipes, traces, session locks, log marks, replay config under `~/.tyto` |
+| `@tyto/cli` | The `tyto` command (`learn`, `run`, `test`, `recipes`, `open`, `brief`, `find`, actions) |
 | `@tyto/compiler` | Coming in slice 6 |
 
 A [YOLOVibeCode](https://github.com/YOLOVibeCode) public repo. Product: Noctusoft, Inc. MIT license.

docs/IMPLEMENTATION.md

@@ -28,9 +28,9 @@ Separate fakes in `@tyto/core/testing`.
 | Port | Shape |
 |---|---|
 | `browser-runner.ts` | `run(argv, opts)`, `batch(steps, opts)` → typed step results; opts carry session, env, AbortSignal |
-| `event-source.ts` | `subscribe(session, signal)` → `AsyncIterable<StreamEvent>` |
+| `browser-events.ts` | `BrowserEventSource.subscribe(session, signal)` → `AsyncIterable<StreamEvent>` |
 | `recipe-store.ts` | `get`, `list`, `save`, `remove` |
-| `trace-store.ts` | `save`, `get`, `latest` |
+| `trace-store.ts` | `save`, `get` |
 | `compiler.ts` | `compile(trace)`, `repair(recipe, misses, passes)` → recipe JSON |
 | `session-lock.ts` | `acquire(session, timeoutMs)` → release function |
 | `log-marks.ts` | `get(session)`, `set(session, counts)`: log offsets taken when a page is opened |
@@ -142,14 +142,32 @@ launch settings change mid-session, and the page is lost (observed: Edge fell ba
   `find locates text anywhere on the page`
 
 ### Slice 5 — recorder and traces
-- `learn opens the session and waits for the sync marker before recording`
+agent-browser's event stream (`<socketDir>/<session>.stream` holds the WebSocket port; socketDir follows
+`AGENT_BROWSER_SOCKET_DIR`, then `$XDG_RUNTIME_DIR/agent-browser`, then `~/.agent-browser`) reports **internal**
+action names (`navigate`, `getbylabel`, `waitforurl`, …), which `actionToArgv` maps back to CLI argv. Its
+`success` field is `false` even for successful commands, so success comes from the result's `error`. Typed values
+arrive in plaintext: the listener holds them in memory only; `tyto learn stop --param input_N=name` keeps chosen
+ones as params. `tyto learn` talks to its detached listener over a unix socket in `~/.tyto/learn/` (dir 0700).
+- `learn waits for the sync marker before recording`
 - `pairs command and result events by id into trace steps`
 - `marks the trace lossy on an orphan result, a missing result, or a socket close`
 - `ignores frame messages`
 - `replaces fill and type values with {{input_N}} placeholders`
 - `keeps only inputs named with --param and drops the rest`
 - `reduces cookie and storage values to names and lengths`
-- `runs the Redactor before the trace is saved`
+- `runs the Redactor before the trace is returned`
+- `actionToArgv maps navigate, click, fill, type, press, select, check, waits, eval, get, snapshot, and find locators`
+- `masks values typed into password-like fields and never offers them as params`
+- `collects the origins the task opened`; `keeps snapshot text and eval results as step output`
+- `StreamEventSource reads the port file and yields parsed events, dropping frames`; `yields a closed event when the socket closes`
+- `agentBrowserSocketDir follows agent-browser's socket directory rules`
+- `FileTraceStore saves traces with mode 0600 and loads them by name`
+- `tyto learn starts the session and a listener, then prints how to finish`; `tyto learn stop requires --task`
+- `tyto learn stop sends the task and kept params, then prints the trace summary`
+- `tyto learn status lists typed inputs by name and locator, never values`
+- `the listener records after the marker and saves the trace with kept params when stopped`
+- `control client and server exchange messages over a private socket`
+- live: `records command/result pairs from the event stream and keeps only the named input`
 
 ### Slice 6 — compiler (Claude Code)
 - `the prompt fences trace page text with a random nonce and labels it as data`

packages/agent-browser/src/index.ts

@@ -1 +1,2 @@
 export { AgentBrowserRunner, type AgentBrowserRunnerOptions } from "./runner.ts";
+export { StreamEventSource, agentBrowserSocketDir, type StreamEventSourceOptions, type WebSocketLike } from "./stream.ts";

packages/agent-browser/src/stream.ts

@@ -0,0 +1,116 @@
+import { readFile } from "node:fs/promises";
+import { homedir } from "node:os";
+import { join } from "node:path";
+import type { BrowserEventSource, StreamEvent } from "@tyto/core";
+
+export type WebSocketLike = {
+  onopen: (() => void) | null;
+  onmessage: ((e: { data: unknown }) => void) | null;
+  onclose: (() => void) | null;
+  onerror: ((e: unknown) => void) | null;
+  close(): void;
+};
+
+export type StreamEventSourceOptions = {
+  socketDir?: string;
+  WebSocketCtor?: new (url: string) => WebSocketLike;
+};
+
+/** agent-browser's `sanitize_session_component`: lowercase alphanumerics, single separators. */
+function sanitize(value: string): string {
+  let out = "";
+  let lastSep = false;
+  for (const c of value) {
+    if (/[\p{L}\p{N}]/u.test(c)) {
+      out += c.toLowerCase();
+      lastSep = false;
+    } else if (out !== "" && !lastSep) {
+      out += c === "_" ? "_" : "-";
+      lastSep = true;
+    }
+  }
+  return out.replace(/[-_]+$/, "");
+}
+
+/** Mirrors agent-browser's `get_socket_dir`, where `<session>.stream` holds the event stream port. */
+export function agentBrowserSocketDir(env: Readonly<Record<string, string | undefined>> = process.env, home = homedir()): string {
+  const explicit = env.AGENT_BROWSER_SOCKET_DIR;
+  const runtime = env.XDG_RUNTIME_DIR;
+  const base = explicit ? explicit : runtime ? join(runtime, "agent-browser") : join(home, ".agent-browser");
+  const ns = env.AGENT_BROWSER_NAMESPACE ? sanitize(env.AGENT_BROWSER_NAMESPACE) : "";
+  return ns ? join(base, "namespaces", ns, "run") : base;
+}
+
+/** Subscribes to a session's WebSocket event stream (lossy by design; frames are dropped). */
+export class StreamEventSource implements BrowserEventSource {
+  readonly #dir: string;
+  readonly #Ws: new (url: string) => WebSocketLike;
+
+  constructor(opts: StreamEventSourceOptions = {}) {
+    this.#dir = opts.socketDir ?? agentBrowserSocketDir();
+    this.#Ws = opts.WebSocketCtor ?? (globalThis.WebSocket as unknown as new (url: string) => WebSocketLike);
+  }
+
+  async *subscribe(session: string, signal: AbortSignal): AsyncIterable<StreamEvent> {
+    if (!/^[\w.-]+$/.test(session)) throw new Error(`invalid session name: ${session}`);
+    let port: string;
+    try {
+      port = (await readFile(join(this.#dir, `${session}.stream`), "utf8")).trim();
+    } catch {
+      throw new Error(`no event stream for session ${session}; is its agent-browser daemon running?`);
+    }
+    if (!/^\d+$/.test(port)) throw new Error(`bad event stream port for session ${session}`);
+
+    const queue: StreamEvent[] = [];
+    let wake: (() => void) | null = null;
+    let done = false;
+    const push = (ev: StreamEvent): void => {
+      queue.push(ev);
+      wake?.();
+    };
+    const ws = new this.#Ws(`ws://127.0.0.1:${port}`);
+    ws.onmessage = (e) => {
+      if (typeof e.data !== "string") return;
+      try {
+        const ev = JSON.parse(e.data) as unknown;
+        if (typeof ev === "object" && ev !== null && !Array.isArray(ev) && (ev as StreamEvent).type !== "frame") push(ev as StreamEvent);
+      } catch {
+        return; // not JSON: the stream is advisory, skip the message
+      }
+    };
+    ws.onclose = () => {
+      if (!done) push({ type: "closed" });
+      done = true;
+    };
+    ws.onerror = () => {
+      if (!done) push({ type: "closed" });
+      done = true;
+    };
+    const onAbort = (): void => {
+      done = true;
+      ws.close();
+      wake?.();
+    };
+    signal.addEventListener("abort", onAbort, { once: true });
+    try {
+      while (!signal.aborted) {
+        const next = queue.shift();
+        if (next) {
+          yield next;
+          if (next.type === "closed") return;
+          continue;
+        }
+        await new Promise<void>((resolve) => {
+          wake = resolve;
+        });
+        wake = null;
+      }
+    } finally {
+      signal.removeEventListener("abort", onAbort);
+      if (!done) {
+        done = true;
+        ws.close();
+      }
+    }
+  }
+}

packages/agent-browser/test/stream.test.ts

@@ -0,0 +1,92 @@
+import { mkdtemp, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, expect, it } from "vitest";
+import type { StreamEvent } from "@tyto/core";
+import { StreamEventSource, agentBrowserSocketDir, type WebSocketLike } from "../src/index.ts";
+
+/** Minimal stand-in for the global WebSocket. */
+class FakeSocket implements WebSocketLike {
+  static last: FakeSocket | null = null;
+  onopen: (() => void) | null = null;
+  onmessage: ((e: { data: unknown }) => void) | null = null;
+  onclose: (() => void) | null = null;
+  onerror: ((e: unknown) => void) | null = null;
+  closed = false;
+  readonly url: string;
+  constructor(url: string) {
+    this.url = url;
+    FakeSocket.last = this;
+    queueMicrotask(() => this.onopen?.());
+  }
+  send(data: unknown): void {
+    this.onmessage?.({ data });
+  }
+  close(): void {
+    this.closed = true;
+    this.onclose?.();
+  }
+}
+
+async function collect(it: AsyncIterable<StreamEvent>, max: number): Promise<StreamEvent[]> {
+  const out: StreamEvent[] = [];
+  for await (const ev of it) {
+    out.push(ev);
+    if (out.length >= max) break;
+  }
+  return out;
+}
+
+describe("StreamEventSource", () => {
+  it("reads the port file and yields parsed events, dropping frames", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "tyto-stream-"));
+    await writeFile(join(dir, "task-1.stream"), "54321\n");
+    const source = new StreamEventSource({ socketDir: dir, WebSocketCtor: FakeSocket });
+    const ac = new AbortController();
+    const pending = collect(source.subscribe("task-1", ac.signal), 2);
+    await new Promise((r) => setTimeout(r, 5));
+    expect(FakeSocket.last?.url).toBe("ws://127.0.0.1:54321");
+    FakeSocket.last?.send(JSON.stringify({ type: "frame", data: "…" }));
+    FakeSocket.last?.send(JSON.stringify({ type: "command", id: "r1", action: "click", params: {} }));
+    FakeSocket.last?.send("not json");
+    FakeSocket.last?.send(JSON.stringify({ type: "result", id: "r1", action: "click" }));
+    const events = await pending;
+    expect(events.map((e) => e.type)).toEqual(["command", "result"]);
+    ac.abort();
+  });
+
+  it("yields a closed event when the socket closes", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "tyto-stream-"));
+    await writeFile(join(dir, "s.stream"), "1234");
+    const source = new StreamEventSource({ socketDir: dir, WebSocketCtor: FakeSocket });
+    const pending = collect(source.subscribe("s", new AbortController().signal), 5);
+    await new Promise((r) => setTimeout(r, 5));
+    FakeSocket.last?.close();
+    expect(await pending).toEqual([{ type: "closed" }]);
+  });
+
+  it("aborting closes the socket and ends the stream", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "tyto-stream-"));
+    await writeFile(join(dir, "s.stream"), "1234");
+    const ac = new AbortController();
+    const pending = collect(new StreamEventSource({ socketDir: dir, WebSocketCtor: FakeSocket }).subscribe("s", ac.signal), 5);
+    await new Promise((r) => setTimeout(r, 5));
+    ac.abort();
+    await pending;
+    expect(FakeSocket.last?.closed).toBe(true);
+  });
+
+  it("a missing port file is a clear error", async () => {
+    const source = new StreamEventSource({ socketDir: await mkdtemp(join(tmpdir(), "tyto-stream-")), WebSocketCtor: FakeSocket });
+    await expect(collect(source.subscribe("nope", new AbortController().signal), 1)).rejects.toThrow(/no event stream for session nope/);
+  });
+});
+
+describe("agentBrowserSocketDir", () => {
+  it("follows agent-browser: AGENT_BROWSER_SOCKET_DIR, then XDG_RUNTIME_DIR/agent-browser, then ~/.agent-browser", () => {
+    expect(agentBrowserSocketDir({ AGENT_BROWSER_SOCKET_DIR: "/s" }, "/home/u")).toBe("/s");
+    expect(agentBrowserSocketDir({ XDG_RUNTIME_DIR: "/run/user/1" }, "/home/u")).toBe("/run/user/1/agent-browser");
+    expect(agentBrowserSocketDir({}, "/home/u")).toBe("/home/u/.agent-browser");
+    expect(agentBrowserSocketDir({ AGENT_BROWSER_NAMESPACE: "team a" }, "/home/u")).toBe("/home/u/.agent-browser/namespaces/team-a/run");
+  });
+});

packages/cli/src/compose.ts

@@ -1,15 +1,24 @@
+import { spawn } from "node:child_process";
 import { homedir } from "node:os";
+import { fileURLToPath } from "node:url";
 import { join } from "node:path";
-import { AgentBrowserRunner } from "@tyto/agent-browser";
-import { SecretRedactor } from "@tyto/core";
-import { FileLogMarks, FileSessionLock, FilesystemRecipeStore, ensureReplayFiles } from "@tyto/store";
+import { AgentBrowserRunner, StreamEventSource } from "@tyto/agent-browser";
+import { Recorder, SecretRedactor } from "@tyto/core";
+import { FileLogMarks, FileSessionLock, FileTraceStore, FilesystemRecipeStore, ensureReplayFiles } from "@tyto/store";
+import { requestControl, serveControl } from "./learn/control.ts";
+import { runListener } from "./learn/listener.ts";
+
+const BIN = fileURLToPath(new URL("../bin/tyto.mjs", import.meta.url));
+const LISTEN_MAX_MS = 2 * 60 * 60 * 1000;
 import type { CliDeps } from "./main.ts";
 
 /** Composition root: the only place real adapters are wired together. */
 export async function composeDeps(env: NodeJS.ProcessEnv = process.env): Promise<CliDeps> {
   const home = env.TYTO_HOME && env.TYTO_HOME !== "" ? env.TYTO_HOME : join(homedir(), ".tyto");
   const paths = await ensureReplayFiles(home);
   const runner = new AgentBrowserRunner({ bin: env.TYTO_AGENT_BROWSER ?? "agent-browser" });
+  const redactor = new SecretRedactor();
+  const socket = (name: string): string => join(home, "learn", `${name}.sock`);
   return {
     store: new FilesystemRecipeStore(join(home, "recipes")),
     exec: {
@@ -20,10 +29,29 @@ export async function composeDeps(env: NodeJS.ProcessEnv = process.env): Promise
     browse: {
       runner,
       marks: new FileLogMarks(join(home, "marks")),
-      redactor: new SecretRedactor(),
+      redactor,
       session: env.AGENT_BROWSER_SESSION && env.AGENT_BROWSER_SESSION !== "" ? env.AGENT_BROWSER_SESSION : "default",
       now: () => Date.now(),
     },
+    learn: {
+      spawnListener: async (name, session) => {
+        const child = spawn(process.execPath, [BIN, "learn", "--listen", name, "--session", session], { detached: true, stdio: "ignore", env });
+        child.unref();
+      },
+      control: (name, req) => requestControl(socket(name), req),
+      listen: async (name, session) => {
+        await runListener(name, session, {
+          recorder: new Recorder({ events: new StreamEventSource(), runner }),
+          traces: new FileTraceStore(join(home, "traces")),
+          redactor,
+          now: () => Date.now(),
+          serve: (handler) => serveControl(socket(name), handler),
+          tracePath: (n) => join(home, "traces", `${n}.json`),
+          maxMs: LISTEN_MAX_MS,
+        });
+        return 0;
+      },
+    },
     out: (line) => process.stdout.write(`${line}\n`),
     err: (line) => process.stderr.write(`${line}\n`),
   };

packages/cli/src/index.ts

@@ -1,2 +1,5 @@
 export { USAGE, main, type CliDeps } from "./main.ts";
 export { composeDeps } from "./compose.ts";
+export { requestControl, serveControl, type ControlReply, type ControlRequest } from "./learn/control.ts";
+export { runListener, type ListenerDeps } from "./learn/listener.ts";
+export { learnCommand, type LearnDeps } from "./learn/commands.ts";