Market Brain Native — Implementation Plan

Execution doc for the spec. Move the market agent's relational brain from Turso into its own Durable Object SQLite, vectors to Vectorize, using Flue/Cloudflare as intended. 2026-06-25

Goal: The market agent reads and writes its 3.1M-fact brain from its own DO's native SQLite (local, no hop), vectors live in Cloudflare Vectorize, cross-agent reads go over Durable Object RPC, and Turso leaves the market path — all on the stable north-fork-valley id, cut over in place behind a flag with the live agent never going dark.

Architecture: Port the injected-client data layer (store.ts/parcels.ts) from the async @libsql/client Client to the synchronous native SqlStorage handle (ctx.storage.sql.exec(sql,…b).toArray()). Vectors move to a [[vectorize]] binding. Brain durability rests on a stable id + an R2 backup: a poisoned session is recovered by recreating the agent (green-field sessions) and re-ingesting from R2 — disposable sessions, re-loadable brain (beta.5 has no in-place session-clear; see Phase A). The one-time ingest runs as a Flue scheduleEvery beat reading NDJSON parts from R2. Cross-agent reads (advisor scout, staff tools) call DO-RPC read methods added to the market DO via extend({base}) — bypassing the harness, so the read never wakes the market model. Markdown (wiki, SOUL.md, AGENT.md, JOURNAL.md) lives as files in the DO via the Workspace file API (Phase F). Convex stays the website's serving/reactive layer, untouched.

Tech: TypeScript · @flue/runtime (beta.5) on Cloudflare Workers/DO · Vectorize · R2 · vitest. Build subagent-driven, Opus on every agent.

Global Constraints — every task inherits these

Flue/CF idioms only. Native ctx.storage.sql (no libsql-impersonating adapter); Vectorize binding for vectors; scheduleEvery+this.state for the ingest; DO-RPC (extend({base}) methods + stub) for cross-agent reads; Workspace over ctx.storage.sql for markdown files. No hand-rolled equivalents of things the framework already does.

The live agent never goes dark. north-fork-valley (currently -v9) stays serving from Turso until native is proven; cutover is in place behind a read-source flag, NOT a version bump. No DO rename anywhere in this plan.

Keep source names. Column/value names stay (fact_value_json, bbox_minx/y/x/y where x=lon y=lat). No renaming 3.1M rows.

Convex mirror untouched. tools.ts:304-309,456-460 Convex writes stay; the website's serving path is unaffected.

Secrets & git. Never print/commit secret values. Branch feature/market-brain-native → PR → squash-merge; never commit to main. The final prod flip + TURSO_* secret drop wait behind Pete's PR review.

Tests. vitest. The native data layer is tested via a synchronous fakeSqlStorage() (backed by node:sqlite/better-sqlite3) that presents .exec(sql,…b).toArray(); existing :memory: tests adapt to it. Vectorize/RPC/R2 bindings are mocked.

Phases

A — Recovery model: disposable sessions, re-loadable brain (superseded keystone)
B — Port the data layer to native SQLite + Vectorize
C — One-time ingest beat (R2 → DO SQLite)
D — Cross-agent reads over DO RPC
E — In-place cutover & cut Turso
F — The native file system: wiki & identity (independent of the cutover)

▢ step✚ write failing test▷ run/verify✓ commit

Phase A — Recovery model (superseded keystone) Built, then demoted. The brain's safety comes from a stable id + R2 backup, not from clearing sessions.

Superseded — 2026-06-25. This phase was originally "clear a poisoned session in place so we never rename." We built it (commits e67d3a8…4d57ea0: clearSessionAction + POST /admin/clear-session, 190 tests green) and then hit a real beta.5 wall: an action runs in an isolated child harness with its own scoped sessions, so harness.sessions.delete('default') clears the action's session, not the live poisoned default — and beta.5 exposes no way to run a turn in a named session to reach it. The frame was wrong anyway. A poisoned session is disposable chat history; the brain is what matters, and it's protected by a stable id (never rename) + an R2 backup (Phase C) that makes a clean recreate cost a few-minute re-ingest, not data loss. So: recovery = recreate the agent + re-ingest from R2. The whole Phase-A machinery (adminActions.ts + the route + the agent wiring) was later removed (commit 7c83e51): it broke flue build --target cloudflare (its file sat in src/agents/ with a camelCase agent name) and couldn't clear the live default session. A2/A3 (version-brick fallback, proof-gate) are moot — nothing gates on in-place clear. Prevention (typed tool outputs + the dream beat) keeps poison rare. Pete: green-field on sessions is fine

A1 · Session-clear action + admin route

Create: src/agents/adminActions.ts · Modify: src/agents/market.ts (register action + handle the dispatch input), src/app.ts (admin route) · Test: test/agents/adminActions.test.ts

Produces: clearSessionAction (a defineAction); POST /admin/clear-session/:agent/:id (auth'd) that dispatch()es {kind:'session-clear', target?} into the instance.

Test: dispatching {kind:'session-clear'} routes to the action and calls harness.sessions.delete('default') (mock harness; assert the delete call + that the action runs in a non-default session).
Write clearSessionAction = defineAction({ name:'admin_clear_session', async run({harness,input}){ await harness.sessions.delete(input?.target ?? 'default'); return {ok:true}; } }).
In market.ts: register the action and wire the agent so a dispatched kind:'session-clear' input invokes it deterministically (not via the model), running in an admin session so it isn't deleting its own session. Confirm the exact dispatch→action wiring against @flue/runtime (the implementer reads the action/dispatch docs).
In app.ts (before app.route('/', flue())): add the auth-gated route that dispatch({agent,id,input:{kind:'session-clear',target}}) and returns 202. Copy the AUTH_PASSWORD gate from market.ts:17-29.
Run: npx vitest run test/agents/adminActions.test.ts → green.
Commit: "feat(market): session-clear action + admin dispatch route".

A2 · Version-brick fallback (honest)

Modify: src/app.ts · Doc: the spec's "rough edge"

Reality: Flue beta.5 exposes NO public DO-surgery surface, and a PersistedSchemaVersionError throws at harness init before any action can run. So an in-place wipe of flue_* on a wedged-at-boot DO isn't cleanly reachable today.

Document the constraint in the route's response + the runbook: if a Flue version bump wedges init, the only levers are (a) settle/wait, or (b) a one-time bin-and-rename — and crucially the brain (our app tables) survives either way; only flue_* session rows are at issue.
Do NOT fabricate a wipe endpoint that can't run. If a future Flue release adds a boot-time session-reset hook, wire it then.
Commit: "docs(market): version-brick recovery constraint + runbook note".

A3 · Proof-gate — clear in place, brain intact

Test: test/agents/sessionClear.proof.test.ts (integration; vitest-pool-workers if available, else a harnessed DO sim)

Gate: the whole native migration is blocked until this passes.

Test: seed a sentinel app table (CREATE TABLE probe; INSERT) in the DO's SQLite; seed/poison a default session; invoke the clear path; assert (1) the session is gone, (2) probe is byte-for-byte intact, (3) the DO id is unchanged.
Run it; confirm green. If SessionBusyError fires, adjust which session the clear runs in until it's reliably non-busy, then re-run.
Commit: "test(market): proof — session clears in place, app tables survive".

Phase B — Port the data layer to native SQLite + Vectorize store.ts + parcels.ts speak ctx.storage.sql directly. Vectors → Vectorize. parcel_embeddings leaves the schema.

B1 · The `SqlStorage` type + `fakeSqlStorage()` test helper

Create: src/market/sqlh.ts (the handle type + tiny helpers), test/helpers/fakeSqlStorage.ts

Produces: type Sql = { exec(q:string, ...b:unknown[]): { toArray(): Array<Record<string,unknown>> } } (matches Flue's SqlStorage, sql-storage-*.d.mts:1-14). fakeSqlStorage(): Sql over a synchronous sqlite (node:sqlite DatabaseSync, else better-sqlite3).

Test: fakeSqlStorage() round-trips exec("CREATE…"); exec("INSERT… VALUES (?)", x); exec("SELECT…").toArray() with correct binding + row shape.
Implement the helper (sync sqlite under a .exec().toArray() facade). No libsql.
Run: npx vitest run test/helpers → green.
Commit: "test: native SqlStorage handle + fakeSqlStorage helper".

B2 · Port `store.ts` (listings/reports) to native SQL

Modify: src/market/store.ts · Test: test/market/store.test.ts

Mechanical transform, every data fn: signature (c: Client, …) → (sql: Sql, …); body await c.execute({sql,args}) → r.rows → sql.exec(query, ...args).toArray(). Functions stay async (call sites unchanged). Drop getTursoClient from this file's hot path (kept only behind the cutover flag — Phase E). Affected: ensureSchema, upsertListing, getListing, getListingBySlug, writeReport, listListings, markBriefed (store.ts:66-277).

Adapt store.test.ts: beforeEach uses fakeSqlStorage() instead of createClient({url:':memory:'}); assertions unchanged.
Transform each fn per the rule above. Preserve SQL text verbatim (it's portable SQLite).
Run: npx vitest run test/market/store.test.ts → green.
Commit: "refactor(store): native ctx.storage.sql data layer (off libsql)".

B3 · Port `parcels.ts` reads; remove `parcel_embeddings`

Modify: src/market/parcels.ts · Test: test/market/parcels.test.ts

Port ensureParcelSchema, getParcel, getParcelFacts, parcelForListing, parcelsByBbox (same transform as B2). Remove the parcel_embeddings CREATE + parcel_emb_idx from ensureParcelSchema (parcels.ts:139-146). Vector fns move to B4.

Adapt parcels.test.ts to fakeSqlStorage(); drop the parcel_embeddings schema assertion; keep parcels/facts/bbox/link tests.
Transform the read fns; delete the embeddings table from the DDL.
Run: npx vitest run test/market/parcels.test.ts → green.
Commit: "refactor(parcels): native SQL reads; drop in-DO embeddings table".

B4 · Vector path → Vectorize binding

Create: src/market/vectors.ts (the Vectorize fns — keeps parcels.ts pure native-SQL) · Modify: src/market/parcels.ts (remove the vector fns + parcel_embeddings), wrangler.jsonc (add [[vectorize]]) · Test: test/market/vector.test.ts — shipped: vector fns landed in a NEW vectors.ts, cleaner than in-place, functionally identical

New signatures: upsertParcelEmbedding(vx: VectorizeIndex, rec:{parcelId,embedding:number[],kind,model?,text?}): Promise<void> → vx.upsert([{id:parcelId, values:embedding, metadata:{kind,market}}]). vectorSearchParcels(vx, sql, queryVec, k, {kind}): Promise<ParcelMatch[]> → vx.query(queryVec,{topK:k,filter:{kind}}) → ids → join to local facts via sql. Same ParcelMatch shape.

Test: mock a Vectorize binding (upsert/query); assert vectorSearchParcels queries Vectorize then joins returned ids to fakeSqlStorage() facts, in score order.
Rewrite the two fns against the Vectorize binding; add "vectorize":[{"binding":"VECTORIZE","index_name":"market-embeddings"}] to wrangler.jsonc.
Run: npx vitest run test/market/vector.test.ts → green. (No writes yet — index stands up empty; on-the-fly writes ride the Phase-2 comps build.)
Commit: "feat(parcels): Vectorize-backed vector search + binding".

B5 · Market agent + its own tools use the native handle

Modify: src/market/tools.ts (sweep/drain/metrics), src/agents/market.ts

Inside the market DO, tools get the handle via getCloudflareContext().storage.sql instead of await store.getTursoClient(). Gate behind the read-source flag (Phase E) so prod still works pre-cutover. Convex mirror calls (tools.ts:304-309,456-460) unchanged.

Test (factory/unit): the market tools resolve their SQL handle from the CF context (mock getCloudflareContext); Convex mirror calls still fire.
Swap the client acquisition; keep every store.*/parcels.* call (signatures now take sql).
Run: npx vitest run test/market → green; npx tsc --noEmit clean.
Commit: "refactor(market): tools read/write the DO's native SQLite".

Phase C — One-time ingest beat (R2 → DO SQLite) Flue scheduleEvery + this.state progress, reading NDJSON parts from R2. Resumable, self-stopping. No bulk-import button — this is the substitute.

C1 · Export the brain to R2 as NDJSON parts

Create: scripts/export-brain-ndjson.mjs (ops, not shipped) · R2: market-brain/north-fork-valley/{parcels,parcel_facts,parcel_listings,pending_resolution}/part-NNNN.ndjson

Source: the cleaned local hz-import.db (or current prod Turso). Each table → line-delimited JSON, chunked (e.g. 20k rows/part). Parsing a 764 MB sqlite binary inside a Worker is impractical; NDJSON range-reads are Worker-native.

Write the exporter (reads sqlite via local sqlite3 -json, writes parts, uploads to R2 with wrangler r2 object put or the R2 API). Record part counts.
Verify: row counts across parts == source (parcels 18,970 · facts 3,114,247 · links 482 · pending 387).
Commit: "ops: brain→R2 NDJSON exporter".

C2 · The ingest beat (extend + scheduleEvery)

Create/Modify: the market agent's cloudflare = extend({base}) block (in src/agents/market.ts or a new src/agents/market.cf.ts), wrangler.jsonc (R2 binding) · Test: test/market/ingest.test.ts

Method async ingestChunk(): read this.state.cursor (table + part index), pull that part from env.BRAIN_BUCKET, INSERT OR IGNORE its rows into getCloudflareContext().storage.sql, advance the cursor in this.state, await this.schedule(3,'ingestChunk') until all tables/parts done, then mark done. Idempotent (re-runnable). Triggered by an auth'd POST /admin/ingest/:id that dispatches the first beat.

Test: mock R2 returning 2 small parts + a fakeSqlStorage(); run the beat to completion; assert rows inserted, cursor advances, re-running is a no-op (OR IGNORE), done set.
Implement the beat + the R2 binding ("r2_buckets":[{"binding":"BRAIN_BUCKET","bucket_name":"market-brain"}]) + the kickoff route.
Run: npx vitest run test/market/ingest.test.ts → green.
Commit: "feat(market): resumable R2→DO ingest beat".

C3 · Count-verify endpoint

Modify: the extend block (add async counts() RPC) or an auth'd read route

Expose row counts per table from the DO's SQLite (auth'd) so cutover can assert exact-match before flipping.
Commit: "feat(market): brain count-verify endpoint".

Phase D — Cross-agent reads over DO RPC Flue has no agent-RPC abstraction; the CF-native mechanism is DO RPC. Market DO exposes read methods; advisor + staff call the stub. Convex untouched.

First step resolves the stub mechanism: the implementer confirms how to obtain a callable stub to the Flue-generated FlueMarketAgent DO from another agent (a declared durable_objects.bindings entry for the class + getByName/idFromName). If a direct stub proves awkward in Flue beta.5, fall back to an auth'd read route on the market agent that runs the query inside the DO and returns JSON. Either way the query executes inside the market DO against its own storage.sql.

D1 · Market DO read methods (RPC)

Modify: the market extend({base}) block, wrangler.jsonc (DO binding for cross-agent calls) · Test: test/market/rpc.test.ts

Methods: async listListingsRpc(limit=60) → store.listListings(getCloudflareContext().storage.sql, limit); async parcelBriefRpc(idOrZpid) → the parcel-brief assembly (parcels.* against the native handle, PII-filtered). Run inside the DO.

Test: invoke the methods against a seeded fakeSqlStorage(); assert listings/brief shapes match today's tool output (PII filtered).
Implement the methods in extend; declare the cross-agent DO binding.
Run: npx vitest run test/market/rpc.test.ts → green.
Commit: "feat(market): DO-RPC read methods (listings, parcel_brief)".

D2 · Re-point staff tools to the stub

Modify: src/staff/marketTools.ts (list_listings, parcel_brief) · Test: test/staff/marketTools.test.ts

Was: const c = await store.getTursoClient(); … store.listListings(c,…). Now: resolve the market DO stub for marketSlug and await stub.listListingsRpc(…) / stub.parcelBriefRpc(…). The toolset shape (the 9-name list the factory test pins) is unchanged.

Test: mock the stub; assert list_listings/parcel_brief call the RPCs and format identically; the factory tool-name list still equals the pinned 9.
Swap the data acquisition; keep formatting.
Run: npx vitest run test/staff → green.
Commit: "refactor(staff): market tools read via DO RPC (off Turso)".

D3 · Re-point advisor scout to the stub

Modify: src/advisor/marketRead.ts (readActiveListings + callers) · Test: test/advisor/marketRead.test.ts

Was: readActiveListings(c: Client,…) over shared env.TURSO_DATABASE_URL. Now: a market-DO-stub read (add a listForSaleRpc(limit) method in D1 if the existing listListingsRpc doesn't carry the FOR_SALE filter). The advisor stops binding Turso entirely.

Test: mock the stub; assert the scout receives the same ScoutListing[] shape.
Re-point; remove the advisor's Turso usage.
Run: npx vitest run test/advisor → green; npx tsc --noEmit clean.
Commit: "refactor(advisor): scout reads market via DO RPC (off Turso)".

Phase E — In-place cutover & cut Turso Same stable id, read-source flag, ingest-then-flip. Final flip + secret drop gated on Pete's PR review. The live agent never goes dark.

E1 · Read-source flag

Modify: src/market/tools.ts, src/market/store.ts client acquisition · Test: exists in B5

MARKET_STORE=turso|native (Worker var). turso → old getTursoClient() path; native → getCloudflareContext().storage.sql. Default turso at deploy so prod is unchanged until we flip.

Wire the flag at the single client-acquisition point; both paths satisfy the same fn signatures (the libsql path via a thin local shim ONLY for the transition, deleted in E4).
Commit: "feat(market): MARKET_STORE read-source flag (default turso)".

E2 · Deploy native code (flag=turso) + run ingest into the live DO

Deploy the branch to homezero-flue with MARKET_STORE=turso. Prod serves from Turso exactly as before; the DO's native SQLite is still empty.
Export brain→R2 (C1), then POST /admin/ingest/north-fork-valley-v9; watch it run to done via the count endpoint (C3). Also copy current live listings + market_reports from Turso into the DO (one-time) so the flip has live data, not just the static brain.
Verify: count endpoint == exact-match (18,970 · 3,114,247 · 482 · 387) + live listings/reports present. Pete reviews the PR here.

E3 · Flip to native + validate

Set MARKET_STORE=native; redeploy. The agent now reads/writes its own SQLite.
Validate live: /home metrics tiles, the Slack desk (parcel_brief, list_listings), a sweep + a metrics beat, and the advisor scout's next read — all green off native. Session-clear route exercised once.
Soak: leave it a night through the cron beats; confirm freshness + no regressions.

E4 · Cut Turso

Modify: store.ts (remove getTursoClient + the transition shim), remove @libsql/client from the market path, wrangler.jsonc

Delete the turso branch of the flag + getTursoClient + the libsql import from the market path. Grep-clean @libsql/client / TURSO_* in the market path → zero.
Run full suite + tsc; deploy. Then drop TURSO_DATABASE_URL/TURSO_AUTH_TOKEN from the Worker (last, after a clean soak).
Commit: "chore(market): remove Turso from the market path".

Phase F — The native file system: wiki & identity A DO has no POSIX disk; its SQLite is the disk. The Workspace file API gives real file I/O over it. Independent of the Turso cutover — buildable any time.

Scope, honestly. This phase stands up the file home only. The rich wiki authoring (interlinked briefs / comps engine) and the dream beat's JOURNAL.md rewrite are existing deferred work — they now have a native place to live, but they are not built here and are not needed for the migration.

F1 · Stand up the Workspace file layer in the market agent

Modify: src/agents/market.ts (or a small src/market/files.ts helper), wrangler.jsonc (the worker_loaders binding only if the shell sandbox is used) · Test: test/market/files.test.ts

Produces: a marketFiles(): WorkspaceFileSystem helper resolving new Workspace({ sql: getCloudflareContext().storage.sql }) (mirror getDefaultWorkspace() in src/sandboxes/cloudflare-shell.ts:232). File ops: read(path), write(path, content), list(prefix) over wiki/*.md, SOUL.md, AGENT.md, JOURNAL.md.

Test (fakeSqlStorage()-backed): write SOUL.md then read it back byte-for-byte; list('wiki/') returns written paths; a second resolve of the handle sees the same files (persistence, not in-memory).
Implement the helper over the Workspace API; seed SOUL.md/AGENT.md with the agent's current persona (lift from src/market/prompts.ts so identity has one home, not two).
Run: npx vitest run test/market/files.test.ts → green; npx tsc --noEmit clean.
Commit: "feat(market): Workspace file layer for wiki + identity (.md)".

F2 · Prove durability live (deploy-time, with E)

After a deploy: write a file via an auth'd probe, force a DO respawn (deploy again), read it back — confirm the file survived. Same durability guarantee as the SQL brain (same storage).
Document in the runbook that wiki/identity files ride the same R2 backup story if a recreate is ever needed (export the file rows alongside the SQL tables).

Self-review checklist

Spec requirement	Covered by
Brain in DO SQLite, native API (no libsql shim)	B1–B3, B5
Vectors in Vectorize, none in the DO	B3 (drop table), B4
Sessions disposable; brain re-loadable (stable id + R2), never rename	A (recovery model), C (R2 backup), E (stable id, in-place flip)
Markdown (wiki, SOUL/AGENT/JOURNAL) lives as DO files via Workspace	F1–F2
Cross-agent reads via DO RPC; Convex untouched	D1–D3; Convex calls untouched in B5
One-time ingest, Flue scheduleEvery, R2	C1–C3
Live agent never dark; in-place cutover; Pete-gated flip	E1–E4
Keep source column/value names	B2–B3 (SQL verbatim)

Execution: subagent-driven, Opus on every agent, fresh subagent per task with a review between. Branch feature/market-brain-native → PR for Pete. Phases A–D + E1/E4-code are branch work; E2/E3 (ingest into prod, the flip) + the secret drop are deploy-time ops gated on Pete's review. Companion to the spec. · 2026-06-25