carlos db region forward, plan 3: pull and pull --at Implementation Plan

> For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Let a member run carlos db pull --out <path> and carlos db pull --at <time> --out <path> against an instance whose data lives in another region: the box ships the planned LTX files, sealed, inside the same single request, and the CLI checks them, restores them locally with today's code and saves the copy, exactly as an in-region pull does.

Architecture: Plans 1 and 2's path with op: "pull". The box never restores for a pull: its worker receives the planned files, verifies each one's LTX checksums and holds them in its private scratch; after check B the agent asks for them back and seals them as the ltx stream in larger chunks. The sealed summary lists the files in order. The console forwards the new op and audits it. The CLI spools the sealed chunks to a private directory rather than memory, checks the whole answer against the summary, writes the files as a litestream file replica, restores them with dbdoor.RestoreLocal (today's restore and lease.CheckDBContext), and hands the copy to today's copyOut and pull lines. The pull's size cap and chunk size are chosen from a measurement on dev before anything is built (Task 1), inside the spec's 22-minute request bound.

Tech Stack: Go 1.26, plans 1 and 2's internal/dbreads, internal/dbdoor, console and CLI code, superfly/ltx's decoder (Verify), litestream's file replica client.

Spec: docs/superpowers/specs/2026-10-22-db-region-forward-design.md (rev 6, which is rev 5 with the plan review's four clarifications; Paul approved building it on 2026-10-22 with the rulings in §16, among them ruling 1: pull goes through the deployment bucket as designed, with the true storage statement, and a per-region bucket is plan 5). Read §2, §5.5 to §5.7, §6, §14 and §16. Plans 1 and 2 must have merged and plan 2 must have rolled to dev.

Branch: db-region-forward-3-pull, cut from amadan main after plan 2 merged.

Global Constraints

Every task's requirements include this section.

• Production-wired acceptance. As plans 1 and 2: the real console server, the real CLI binary through carlosTestEnv with a scratch cmd.Dir, real stores (MinIO in CI), real verify, seal and restore code. The in-process box (dbreadstest) is not acceptance for the units; plan 1's root-runner tests are, and Task 5 extends them for the pull, once the agent can make one. Scripted boxes are stand-ins and say so.
• Targeted local runs; CI runs the full suite. Never make test or a whole console, host or root package locally; build, vet, gofmt and the cross-compiles run locally.
• Cross-compile gate. GOOS=darwin GOARCH=arm64 go build ./... and GOOS=windows GOARCH=amd64 go build ./... before every push. The CLI's pull path is portable.
• Copy needs review before it ships, with no em dashes, lower case after a full stop (ruling 4). Task 2 reviews the pull rows; no pull string reaches code first.
• No tenant names. shop, shop.example.com, far.example.com, member@example.com, regions ie and uk.
• Go cache at its defaults; the sandbox off for go commands if the cache is read-only, and for socket tests and binary builds.
• Census overlays per branch, loaded LAST after whatever main loads last at the merge (plan 2's db-region-forward-2 at the earliest), previous seal first; rows rev3=PROPOSED-db-region-forward-3; ruling asked on #280.
• Merge main first before review and before merging; merge only behind git merge-base --is-ancestor origin/main HEAD.
• Test weights for new console or root tests of a second or more.
• No CI-only scripts under scripts/; MinIO additions go in .amadan/ci.d/30-minio-contract.
• New host tests call t.Parallel() unless they touch env, cwd, umask or package vars.
• Scratch under $TMPDIR or t.TempDir(), removed by name. The CLI's own spool directory is under os.TempDir(), 0700, removed on every path, as today's restore directory is.
• Ask before anything large. A test or measurement that writes more than a few GB (a full-size pull) runs only on Paul's say-so (machine rule 5, the CPU budget note).
• Wire shapes are additive. The forward marker gains "pull" in forward_ops; a CLI from plan 2 still prints DBF-8 for a pull.
• Spec numbers, verbatim, as plans 1 and 2, and: the request at most 22 minutes, the CLI's context 23; the pull's cap and chunk size are Task 1's, recorded in the spec before use.

Review Focus

The five input classes the spec implies but no gate test names, most likely first. Each has its test in the owning task.

1. The member's disk fills mid-pull. A pull spools up to the cap of ciphertext, then writes the files and the restored copy: about three times the cap under os.TempDir(). A person expects a clean failure, nothing at --out, and the spool directory gone. Test: Task 7, TestForwardedPullLeavesNothingWhenTheSpoolFails (through dbPull, the third spool write failing).
2. A planned file that is corrupt in the bucket. A person expects the box's worker to refuse it (check-failed, today's sentence), never to ship it, and never a half-written --out. Test: Task 4, TestWorkerHoldRefusesACorruptFile.
3. Files that do not add up. A summary whose file sizes do not sum to its byte count, or a file list that does not end at the summary's transaction. A person expects DBF-13 and nothing at --out. Test: Task 7, TestForwardedPullRefusesFilesThatDoNotAddUp.
4. Ctrl-C during a long pull. A person expects stopped, nothing at --out, and the spool directory gone. Tests: Task 7, TestForwardedPullCtrlCLeavesNothing (through dbPull), and Task 9, TestDBForwardPullInterruptedLeavesNothing (SIGINT to the binary).
5. A box whose cap is raised past the CLI's bound. CARLOS_DB_PULL_MAX_BYTES above the built-in cap would ship more than any CLI accepts. A person expects the box to clamp to the built-in cap. Tests: Task 5, TestPullCapOnlyLowers and TestNewAgentClampsMaxPull.

Plan review, round 1

Astra's findings (docs/superpowers/specs/CODEX-FINDINGS-DB-REGION-FORWARD-PLANS.md) for this plan, and where each landed:

• 8 (the installed-unit pull test came before the agent's pull): it is Task 5's Step 5 now; Task 4 keeps the worker's own tests.
• 21 (the measurement needed telemetry nobody built): plan 1's agent logs its planning-and-restore time and its confirmed output, plan 2's console logs its relay; Task 1 measures from those two lines.
• 22 (malformed sizes could panic the CLI): checkPullManifest checks the whole file list before any file is created. Tests: four new malformed cases, each asserting DBF-13, no panic and no file.
• 23 (the cleanup tests did not run cleanup): the spool failure and Ctrl-C run dbPull itself against a stand-in console, with TMPDIR scratch and a spool-write seam; Task 9 sends SIGINT to the binary mid-pull.
• 24 (mutations that could not turn their tests red): TestNewAgentClampsMaxPull calls NewAgent; the malformed-manifest cases assert no file is created; the record-changed case asserts the worker never received run.

None declined.

Plan review, round 2

From CODEX-FINDINGS-DB-REGION-FORWARD-PLANS-R2.md:

• 26 (Task 5 dropped plan 1's timing lines and left attemptStart unused): the replacement keeps both lines, the first saying held through for a pull. TestPullShipsThePlannedFiles asserts them.
• 32 (the hash test failed before the hash check): the scripted answer is far's real planned files with a valid manifest; only the digest is wrong.
• 37 (the spool's growth): readDBForward takes plan 2's dbfChunkBound and checks it before the sink keeps anything. Test: TestForwardedPullSpoolsNoMalformedOrExcessChunk.
• The Ctrl-C test now holds after lines[:3] (admitted, picked, chunk 1); lines[:2] would never have spooled a chunk.

None declined.

Census impact

• Plan 14 root-read census: one overlay, plan14-root-reads-db-region-forward-3.tsv, rev3=PROPOSED-db-region-forward-3: the worker's pull units (receive with LTX verification, send back) at J4 as the db-read worker; the agent's changed read units re-frozen at plan 1's dispositions with the pull appended; the console's changed reads-route units re-frozen at plan 2's; new CLI units for the pull (the spool directory's writes and reads, the file replica it writes, the restore) at J4 as the carlos CLI; envelopes ., internal/dbreads, internal/console, internal/host (for the root-runner dispatch, tests only, so only if the census names it). The seal moves.
• 9B8 record-write census: no new write site is expected (the pull reuses plan 1's chunk writer and plan 2's request writes); a changed digest of an existing nonrecord or key-site entry is re-pinned at its class.
• Claim routes: unchanged.
• go.mod: unchanged.

Copy strings

Drafted here for Task 2's review; none reaches code before Paul approves it:

| id | file | text | |---|---|---| | DBF-17 | dbforward.go | <host>'s database is <size> of replica files. carlos db pull can fetch at most <cap> from region <region>. | | DBF-18 | dbforward.go | fetching a copy of <host>'s database from region <region>. | | DBF-19 | internal/console/apidbforward.go | forwarded a database pull to region <region> (cli) | | DBF-20 | internal/console/apidbforward.go | forwarded a database pull to region <region>, as of <time> (cli) | | DBF-23 | docs/site/cli.md | If the app has more than one instance, name one with --instance. Only instances that sleep have a replica. An instance on your own fleet or in a pool cannot be read this way, and the command says which.\n\nWhen an instance keeps its data in another region than the console's, carlos db query runs in that region. A server there restores the copy, runs your query and sends back only the rows. carlos db pull fetches the copy's replica files from there and restores them on your machine. Both are encrypted for your terminal on the way, so nothing in between can read them, and nothing is printed or saved until all of it has arrived and checked out. There, a query's copy can be at most 2 GB once restored, the query runs for at most 5 minutes, and the rows can come to at most 64 MB. A pull can fetch at most <pull cap> of replica files. | | DBF-24 | dbforward.go | the server in region <region> does not have room to prepare a copy of <host>'s database right now. try again later. |

<pull cap> in DBF-23 is Task 1's number, written in before the review (for example 2 GB). DBF-23 replaces DBF-30 in the CLI page; DBF-30 stays in the record, approved, and leaves the built list. DBF-8 stays built: a console from plan 2 does not forward pulls, and a new CLI still says DBF-8 for it. Reused unchanged: today's three pull lines ("Saved a copy of ...", dbPullHolds, "It is a copy. ..."), the dbRestoreError sentences, stopped, and plan 2's DBF-3, DBF-12, DBF-13, DBF-15 and DBF-16.

File structure

| File | Responsibility | |---|---| | internal/dbreads/limits.go, envelope.go, lines.go, chunks.go (modify) | MaxPullBytes, PullChunkBytes, MaxLineBytes; pull accepted; the chunk size a parameter. | | internal/dbreads/worker.go (modify) | The worker's hold mode: receive, verify, hold, send back. | | internal/dbreads/read.go, agent.go (modify) | The agent's pull: the cap, the reservation, the ltx stream, the summary's files. | | internal/host/dbreads.go (modify) | CARLOS_DB_PULL_MAX_BYTES, lowering only. | | internal/console/apidbforward.go (modify) | forward_ops gains pull; the route admits it; DBF-19, DBF-20. | | dbforward.go, db.go (modify) | The CLI's spool, its pull opener, dbPullForward; DBF-17, DBF-18, DBF-24. | | docs/site/cli.md (modify) | DBF-23. | | Tests: internal/dbreads/pull_test.go, internal/console/apidbforward_test.go, internal/console/dbforward_minio_test.go, dbforward_test.go, dbforward_e2e_test.go, internal/host/dbreads_rootrunner_linux_test.go (create or modify) | The pull's unit, acceptance and root-runner tests. |

---

Task 1: Measure, then choose the pull's cap and chunk size

Spec §14: "A 2 GiB pull in one request is the case most likely to need a longer request bound; plan 3 measures it before choosing." The choice rule here keeps the request bound (22 minutes, J1) and sizes the pull to fit it, because a longer request would reopen every timing argument of spec §5.2.

Files:

• Modify: docs/superpowers/specs/2026-10-22-db-region-forward-design.md (§5.7's limits table and a short rev 7 note)

• [ ] Step 1: Measure the forwarded path's throughput on dev

On dev, after plan 2 has rolled and CARLOS_DB_FORWARD=1 is set on the dev console, from a laptop signed in to it, against an instance in dev's uk region:

time carlos db query --app <dev app> --instance <its uk instance> \
  --sql "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c LIMIT 32000) SELECT hex(randomblob(1000)) FROM c" > "$TMPDIR/dbf-measure.out"
wc -c "$TMPDIR/dbf-measure.out"; rm "$TMPDIR/dbf-measure.out"

That is about 64 MB of rows (the output cap; the command ends with DBF-6, which is expected). Run it three times. Each run leaves two lines that plans 1 and 2 write for this measurement (plan review finding 21):

• the box's, in journalctl -u carlos-edge on the uk box (read through the operator's usual access): dbreads: <id>: output <B> bytes in <N> confirmed chunks in <T_box>, the time from run to the last chunk's confirmed create, which excludes planning and restore (the line before it, planned ... and restored to ... in ..., is the restore's; a pull's says held through);
• the console's, in journalctl -u console: console: db read <id>: relayed <N> chunks (<B'> bytes) in <T_console>, ended <why>, the whole stream from the 200 to its last line.

The box's rate is B / Tbox. The console's is B' / (Tconsole − Tplan), where Tplan is the planned ... in <T_plan> line's duration; that leaves the pickup inside the console's time, so it understates the console's rate, which only makes the cap smaller. R is the slower of the two, in MB/s. Record all three runs, with both lines of each.

• [ ] Step 2: Apply the rule

• The pull's transfer must fit the owner's deadline with room for planning: cap = floor(R × 15 minutes / 256 MiB) × 256 MiB, at most 2 GiB (ruling 2's restored-size cap: a pull never needs more replica bytes than the database it restores, for a compacted replica).
• The chunk size: 1 MiB (PullChunkBytes), so a 2 GiB pull is 2,048 objects, not 8,192. Each object is a PUT on the box and a GET on the console, and their per-request latency, not their bytes, bounds a small chunk's rate; a larger chunk only helps. Keep 1 MiB.
• dbreads.MaxLineBytes becomes 2 MiB, which a 1 MiB chunk's sealed, base64 line (about 1.4 MB) fits.

• [ ] Step 3: Record it in the spec

In §5.7's limits table add, under "Output":

| Pull | at most <cap> of planned LTX files (measured on dev, <date>: relay rate <R> MB/s); `CARLOS_DB_PULL_MAX_BYTES` on a box can only lower it; the read ends `too-large` | agent, from the plan, before any byte moves; the CLI's spool is bounded by the same number |
| Pull chunk | 1 MiB or 1 s, whichever comes first | agent |

and a line in the status header: "Rev 7 (plan 3, <date>): the pull's cap and chunk size, from a measurement on dev; the request bound is unchanged." Ask Paul to confirm the cap as a limit (ruling 2 covers the others), and write his answer into §16's ruling 2. Commit:

git add docs/superpowers/specs/2026-10-22-db-region-forward-design.md
git commit -m "db-region-forward spec rev 7: the pull's cap and chunk size, measured on dev"

Every later task uses the recorded numbers. Where this plan writes 2 << 30 for the cap, use Task 1's value if it is lower.

---

Task 2: The pull rows: draft, review, record

Files:

• Modify: docs/superpowers/plans/2026-10-22-db-region-forward-copy.md
• Modify: approved_db_forward_copy_test.go

• [ ] Step 1: Add the rows, pending

Add a "Pull (plan 3)" table to the record with DBF-17, DBF-18, DBF-19, DBF-20, DBF-23 and DBF-24 as drafted in this plan's copy strings, each pending, with Task 1's cap written into DBF-23. In the gate, extend dbForwardCopyIDs:

var dbForwardCopyIDs = func() []string {
	ids := make([]string, 0, 27)
	for i := 1; i <= 24; i++ {
		if i == 21 || i == 22 {
			continue // appended below, in the record's order
		}
		ids = append(ids, "DBF-"+strconv.Itoa(i))
	}
	return append(ids, "DBF-21", "DBF-22", "DBF-30", "DBF-31", "DBF-32")
}()

The list is DBF-1 to DBF-20, DBF-23, DBF-24, then DBF-21, DBF-22, DBF-30, DBF-31 and DBF-32; checkMovesInventory compares the sets. Run go test . -run 'TestDBForwardCopy' -count=1. Expected: PASS.

• [ ] Step 2: Review

Invoke the copy-review skill with the six rows, with context (DBF-17: a pull refused as too big; DBF-18: stderr while a pull waits; DBF-19, DBF-20: the audit row; DBF-23: the CLI page's paragraph, replacing DBF-30; DBF-24: a pull refused for want of disk on the box). Apply Paul's text verbatim, mark the rows approved, and run the gate again.

• [ ] Step 3: Commit

git add docs/superpowers/plans/2026-10-22-db-region-forward-copy.md approved_db_forward_copy_test.go
git commit -m "db forward: the pull rows, approved"

---

Task 3: The wire: pull in the envelope, the pull's limits, the chunk size

Files:

• Modify: internal/dbreads/limits.go, internal/dbreads/envelope.go:39 (ops) and valid(), internal/dbreads/lines.go (MaxLineBytes), internal/dbreads/chunks.go (newChunkWriter takes a chunk size), internal/dbreads/read.go (its one newChunkWriter call)
• Modify: internal/dbreads/envelope_test.go, internal/dbreads/read_test.go (the newChunkWriter call in TestChunkWriterCapIsExact)

Interfaces:

• Produces: dbreads.MaxPullBytes int64 = 2 << 30 (or Task 1's cap), dbreads.PullChunkBytes = 1 << 20, dbreads.MaxLineBytes = 2 << 20; newChunkWriter(ctx, s, id, pub, stream string, every time.Duration, max int64, size int) *chunkWriter; Request.valid accepts op: "pull" with an empty sql.

• [ ] Step 1: Write the failing tests

Append to internal/dbreads/envelope_test.go:

func TestEnvelopeCarriesAPull(t *testing.T) {
	t.Parallel()
	pub, priv := rootKeys(t)
	req := testRequest(t)
	req.Op, req.SQL = OpPull, ""
	wire, err := Sign(release.RootSigner(priv), req)
	if err != nil {
		t.Fatal(err)
	}
	if got, err := Verify(pub, wire, req.ID, "uk", testNow); err != nil || got.Op != OpPull {
		t.Fatalf("%+v %v", got, err)
	}
	req.SQL = "SELECT 1"
	if _, err := Sign(release.RootSigner(priv), req); !errors.Is(err, ErrEnvelope) {
		t.Fatalf("a pull carrying SQL signed: %v", err)
	}
}

and to internal/dbreads/lines_test.go:

func TestAFullPullChunkLineFits(t *testing.T) {
	t.Parallel()
	full, _ := json.Marshal(Line{Chunk: 1, Data: bytes.Repeat([]byte{0xff}, PullChunkBytes+SealOverhead)})
	if len(full)+1 > MaxLineBytes {
		t.Fatalf("a full pull chunk line is %d bytes", len(full))
	}
}

• [ ] Step 2: Run them to verify they fail

Run: go test ./internal/dbreads/ -run 'TestEnvelopeCarriesAPull|TestAFullPullChunkLineFits' -count=1 Expected: FAIL (PullChunkBytes undefined; a pull refused).

• [ ] Step 3: Make the changes

In limits.go add:

	// MaxPullBytes is a pull's cap on its planned LTX files (spec §5.7, rev
	// 6). CARLOS_DB_PULL_MAX_BYTES on a box can only lower it, so no box
	// ships more than a CLI accepts.
	MaxPullBytes int64 = 2 << 30
	// PullChunkBytes is a pull's chunk: larger than a query's, so a full
	// pull is about two thousand objects.
	PullChunkBytes = 1 << 20

In lines.go, MaxLineBytes = 2 << 20 with its comment saying a pull chunk's line is about 1.4 MB. In envelope.go, var ops = map[string]bool{OpQuery: true, OpPull: true} and in valid():

	switch r.Op {
	case OpQuery:
		if r.SQL == "" || len(r.SQL) > MaxSQLBytes || !utf8.ValidString(r.SQL) {
			return false
		}
	case OpPull:
		if r.SQL != "" {
			return false
		}
	}

replacing the query-only SQL check. In chunks.go, add a size int field and parameter to chunkWriter/newChunkWriter and use c.size where it used ChunkBytes; plan 1's query call passes ChunkBytes, and TestChunkWriterCapIsExact passes ChunkBytes.

• [ ] Step 4: Run the package

Run: go test ./internal/dbreads/ -count=1 -race Expected: PASS.

• [ ] Step 5: Commit

git add internal/dbreads
git commit -m "dbreads: pull in the envelope, the pull's cap and chunk size"

---

Task 4: The worker's hold mode

Files:

• Modify: internal/dbreads/worker.go
• Create: internal/dbreads/pull_test.go (this task's tests)

The installed-unit pull test needs the agent's pull and is Task 5's (plan review finding 8).

Interfaces:

• Consumes: receiveFile, frameWriter, the frames (plan 1 Task 7); ltx.NewDecoder, (*ltx.Decoder).Verify.
• Produces: ServeWorker serves planHeader{Op: OpPull}: receive every file, verify it, answer ready, then on run send every file back in the header's order as output frames and end; on stop, nothing.

• [ ] Step 1: Write the failing tests

Create internal/dbreads/pull_test.go (net.Pipe to the real worker code: not acceptance):

package dbreads

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
)

// sendPullPlan sends a pull header and the planned files; it returns the
// files' bytes concatenated, which is what the worker must send back.
func sendPullPlan(t *testing.T, w io.Writer, dir string, corrupt bool) (dbdoor.RestorePlan, []byte) {
	t.Helper()
	rd := lsfile.NewReplicaClient(dir)
	plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	if err := writeJSONFrame(w, frameHeader, planHeader{V: ProtocolVersion, Op: OpPull, Files: plan.Planned, Target: plan.TXID, MaxRestored: DefaultMaxRestored}); err != nil {
		t.Fatal(err)
	}
	var all bytes.Buffer
	for i, f := range plan.Planned {
		rc, _ := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		b, _ := io.ReadAll(rc)
		rc.Close()
		if corrupt && i == 0 {
			b[len(b)/2] ^= 0xff
		}
		all.Write(b)
		writeFrame(w, frameFile, b)
	}
	return plan, all.Bytes()
}

func TestWorkerHoldsThenSendsThePlannedFiles(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	conn, done := workerPipe(t, WorkerOptions{})
	plan, want := sendPullPlan(t, conn, dir, false)
	var ready readyMsg
	readTyped(t, conn, frameReady, &ready)
	if ready.TXID != plan.TXID {
		t.Fatalf("ready at %d", ready.TXID)
	}
	writeJSONFrame(conn, frameRun, runMsg{})
	var got bytes.Buffer
	for {
		typ, p, err := readFrame(conn)
		if err != nil {
			t.Fatal(err)
		}
		if typ == frameOutput {
			got.Write(p)
			continue
		}
		var end endMsg
		if typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != "" {
			t.Fatalf("end %q %s", typ, p)
		}
		break
	}
	if !bytes.Equal(got.Bytes(), want) {
		t.Fatalf("sent back %d bytes, held %d", got.Len(), len(want))
	}
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

// Review Focus 2: a corrupt file is refused before ready, never shipped.
func TestWorkerHoldRefusesACorruptFile(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 2)
	conn, done := workerPipe(t, WorkerOptions{})
	sendPullPlan(t, conn, dir, true)
	var end endMsg
	readTyped(t, conn, frameEnd, &end)
	if end.Code != CodeCheckFailed {
		t.Fatalf("end = %+v", end)
	}
	<-done
}

func TestWorkerHoldStopsWithoutSending(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 1)
	conn, done := workerPipe(t, WorkerOptions{})
	sendPullPlan(t, conn, dir, false)
	readTyped(t, conn, frameReady, nil)
	writeFrame(conn, frameStop, nil)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

• [ ] Step 2: Run them to verify they fail

Run: go test ./internal/dbreads/ -run 'TestWorkerHold' -count=1 Expected: FAIL: the worker refuses op: pull as a protocol error.

• [ ] Step 3: Write the hold mode

In ServeWorker, accept both ops (h.Op != OpQuery && h.Op != OpPull is the protocol error), and after the files are received branch before the restore:

	if h.Op == OpPull {
		return holdAndSend(ctx, conn, next, end, root, h)
	}

and add:

// holdAndSend is a pull's whole worker job (spec §6: the box never restores
// for a pull). Each planned file's LTX checksums are verified (the decode
// is the worker's, as every decode of tenant bytes is), the files are held
// in this unit's private scratch, and after the agent's check B they go
// back in the header's order, whole, as output.
func holdAndSend(ctx context.Context, conn io.Writer, next func() (frame, error), end func(string, error) error, root string, h planHeader) error {
	paths := make([]string, len(h.Files))
	for i, lf := range h.Files {
		paths[i] = litestream.LTXFilePath(root, lf.Level, ltx.TXID(lf.MinTXID), ltx.TXID(lf.MaxTXID))
		if err := verifyLTX(paths[i]); err != nil {
			return end(CodeCheckFailed, err)
		}
	}
	if len(h.Files) == 0 || h.Files[len(h.Files)-1].MaxTXID != h.Target {
		return end(CodeCheckFailed, errors.New("the plan does not end at its target"))
	}
	if err := writeJSONFrame(conn, frameReady, readyMsg{TXID: h.Target}); err != nil {
		return err
	}
	f, err := next()
	if err != nil {
		return err
	}
	switch f.typ {
	case frameStop:
		return nil
	case frameRun:
	default:
		return errProtocol
	}
	out := &frameWriter{w: conn}
	for _, p := range paths {
		if ctx.Err() != nil {
			return errAgentGone
		}
		in, err := os.Open(p)
		if err != nil {
			return end(CodeCheckFailed, err)
		}
		_, err = io.Copy(out, in)
		in.Close()
		if err != nil {
			return err
		}
	}
	if err := out.flush(); err != nil {
		return err
	}
	return end("", nil)
}

// verifyLTX decodes one held file whole and checks its checksums.
func verifyLTX(path string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	return ltx.NewDecoder(f).Verify()
}

• [ ] Step 4: Run the unit tests

Run (sandbox off): go test ./internal/dbreads/ -run 'TestWorker' -count=1 -race Expected: PASS.

• [ ] Step 5: Commit

git add internal/dbreads
git commit -m "dbreads: the worker's hold mode for a pull: verify, hold, send back after check B"

---

Task 5: The agent's pull

Files:

• Modify: internal/dbreads/read.go (attempt branches on the op), internal/dbreads/agent.go (AgentConfig.MaxPull)
• Modify: internal/host/dbreads.go (CARLOS_DB_PULL_MAX_BYTES)
• Modify: internal/dbreads/pull_test.go, internal/dbreads/fixture_test.go (plan 1's: inProcWorkers records the frames the agent sends), internal/host/dbreads_test.go
• Modify: internal/host/dbreads_rootrunner_linux_test.go (a pull through the installed units, after the agent can make one: plan review finding 8)

Interfaces:

• Consumes: attempt, newChunkWriter with a size (Task 3), relay, ended, the worker's hold mode (Task 4).
• Produces: AgentConfig.MaxPull int64 (MaxPullBytes when zero or larger); dbReadsEnv.maxPull; a pull's result: done with chunks of the ltx stream and a summary carrying TXID, At, Files (the plan, in order), Bytes and SHA256; too-large with Restored (the plan's LTX bytes) and Cap; no-room.

• [ ] Step 1: Write the failing tests

Append to internal/dbreads/pull_test.go, and add "fmt", "strings" and "sync" to its imports:

func (fx *fixture) pullRequest(at string) Request {
	req := fx.request("", at)
	req.Op = OpPull
	return req
}

// answerPull opens a pull's chunks as the CLI does: the ltx stream.
func (fx *fixture) answerPull(id string) ([]byte, Summary, Result) {
	fx.t.Helper()
	raw, _, err := fx.store.Get(context.Background(), ResultKey(id))
	if err != nil {
		fx.t.Fatal(err)
	}
	var r Result
	json.Unmarshal(raw, &r)
	s, err := OpenSummary(fx.reply, id, r.Sealed)
	if err != nil {
		fx.t.Fatal(err)
	}
	var out bytes.Buffer
	for seq := int64(1); seq <= r.LastSeq; seq++ {
		sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(id, seq))
		plain, err := OpenChunk(fx.reply, id, seq, StreamLTX, seq == r.LastSeq, sealed)
		if err != nil {
			fx.t.Fatalf("chunk %d: %v", seq, err)
		}
		out.Write(plain)
	}
	return out.Bytes(), s, r
}

func TestPullShipsThePlannedFiles(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 4)
	var logMu sync.Mutex
	var logs []string
	a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
		c.Logf = func(format string, args ...any) {
			logMu.Lock()
			defer logMu.Unlock()
			logs = append(logs, fmt.Sprintf(format, args...))
		}
	})
	req := fx.pullRequest("")
	fx.submit(req)
	a.Pass(context.Background())
	a.Wait()
	got, s, r := fx.answerPull(req.ID)
	plan, _ := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx.replica), time.Time{})
	var want bytes.Buffer
	for _, f := range plan.Planned {
		rc, _ := lsfile.NewReplicaClient(fx.replica).OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		io.Copy(&want, rc)
		rc.Close()
	}
	if r.State != StateDone || !bytes.Equal(got, want.Bytes()) || s.TXID != plan.TXID || len(s.Files) != len(plan.Planned) || s.Bytes != plan.LTXBytes {
		t.Fatalf("result %+v summary %+v: %d bytes, want %d", r, s, len(got), want.Len())
	}
	for seq := int64(1); seq < r.LastSeq; seq++ {
		sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(req.ID, seq))
		if len(sealed)-SealOverhead > PullChunkBytes {
			t.Fatalf("chunk %d is %d bytes", seq, len(sealed))
		}
	}
	// Task 1's measurement reads these two lines (plan review finding 26).
	logMu.Lock()
	all := strings.Join(logs, "\n")
	logMu.Unlock()
	if !strings.Contains(all, "and held through") || !strings.Contains(all, "confirmed chunks in") {
		t.Fatalf("the pull's timing lines are missing:\n%s", all)
	}
}

func TestPullTooLargeAndNoChunkBeforeCheckB(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 2)
	a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxPull = 100 })
	req := fx.pullRequest("")
	fx.submit(req)
	a.Pass(context.Background())
	a.Wait()
	_, s, r := fx.answerPull(req.ID)
	if r.Code != CodeTooLarge || s.Cap != 100 || s.Restored <= 100 {
		t.Fatalf("%+v %+v", r, s)
	}
	// T9 for a pull: a record change after the last byte left the bucket.
	fx2 := newFixture(t, 2)
	plan, _ := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx2.replica), time.Time{})
	last := plan.Planned[len(plan.Planned)-1]
	hook := &stageHook{ReplicaReader: lsfile.NewReplicaClient(fx2.replica), stage: "after the last byte",
		lastSpan: dbdoor.Span(last.MinTXID, last.MaxTXID), change: func() { r := fx2.record(); r.Region = "ie"; fx2.putRecord(r) }}
	b, w := fx2.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
		c.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return hook, nil }
	})
	req2 := fx2.pullRequest("")
	fx2.submit(req2)
	b.Pass(context.Background())
	b.Wait()
	if _, _, r := fx2.answerPull(req2.ID); r.Code != CodeRecordChanged || len(fx2.chunkKeys(req2.ID)) != 0 {
		t.Fatalf("%+v, chunks %v", r, fx2.chunkKeys(req2.ID))
	}
	// The worker never heard run: check B failed, so the agent said stop
	// (plan review finding 24). Without this, an agent that sent run first
	// and refused the files afterwards would pass the two checks above.
	for _, typ := range w.sentFrames() {
		if typ == frameRun {
			t.Fatalf("the agent sent run before check B passed: %v", w.sentFrames())
		}
	}
}

// Plan review finding 24: NewAgent itself clamps the pull's cap, whatever a
// box's settings hand it.
func TestNewAgentClampsMaxPull(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	for give, want := range map[int64]int64{0: MaxPullBytes, -1: MaxPullBytes, 1 << 20: 1 << 20, MaxPullBytes + 1: MaxPullBytes} {
		a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxPull = give })
		if a.cfg.MaxPull != want {
			t.Errorf("MaxPull %d became %d, want %d", give, a.cfg.MaxPull, want)
		}
	}
}

In plan 1's internal/dbreads/fixture_test.go, let inProcWorkers record the type of every frame the agent writes. writeFrame writes each frame in a single Write, so the first byte of each Write on the agent's end is a frame type:

// frameTap records the first byte of each Write the agent makes: one frame
// type per frame (writeFrame writes a frame in one Write).
type frameTap struct {
	net.Conn
	w *inProcWorkers
}

func (f frameTap) Write(p []byte) (int, error) {
	if len(p) > 0 {
		f.w.mu.Lock()
		f.w.sent = append(f.w.sent, p[0])
		f.w.mu.Unlock()
	}
	return f.Conn.Write(p)
}

func (w *inProcWorkers) sentFrames() []byte {
	w.mu.Lock()
	defer w.mu.Unlock()
	return append([]byte(nil), w.sent...)
}

with sent []byte added to inProcWorkers beside liveAtDial, and both of dial's return agentEnd, nil lines becoming return frameTap{Conn: agentEnd, w: w}, nil.

Append to internal/host/dbreads_test.go:

// Review Focus 5: the pull's cap can only be lowered on a box.
func TestPullCapOnlyLowers(t *testing.T) {
	t.Parallel()
	pub, _, _ := ed25519.GenerateKey(rand.Reader)
	base := map[string]string{"CARLOS_HIBERNATE_BUCKET": "d", "CARLOS_REGION": "uk", "CARLOS_RELEASE_PUBKEY": base64.StdEncoding.EncodeToString(pub)}
	get := func(v string) func(string) string {
		return func(k string) string {
			if k == "CARLOS_DB_PULL_MAX_BYTES" {
				return v
			}
			return base[k]
		}
	}
	if env, _ := dbReadsSettings(get("")); env.maxPull != dbreads.MaxPullBytes {
		t.Fatalf("default %d", env.maxPull)
	}
	if env, _ := dbReadsSettings(get("1048576")); env.maxPull != 1<<20 {
		t.Fatalf("lowered %d", env.maxPull)
	}
	if env, _ := dbReadsSettings(get("99999999999999")); env.maxPull != dbreads.MaxPullBytes {
		t.Fatalf("raised past the built-in cap: %d", env.maxPull)
	}
}

• [ ] Step 2: Run them to verify they fail

Run (sandbox off): go test ./internal/dbreads/ -run 'TestPull|TestNewAgentClampsMaxPull' -count=1 && go test ./internal/host/ -run 'TestPullCapOnlyLowers' -count=1 Expected: FAIL.

• [ ] Step 3: Write the agent's pull

In agent.go, add MaxPull int64 to AgentConfig (test seam and box setting) and in NewAgent:

	if cfg.MaxPull <= 0 || cfg.MaxPull > MaxPullBytes {
		cfg.MaxPull = MaxPullBytes
	}

In read.go's attempt, replace plan 1's restored-size check and its a.adm.grow call (the two statements right after PlanFrom's error switch) with a branch on the op for the cap and the reservation:

	pull := req.Op == OpPull
	switch {
	case pull && plan.LTXBytes > a.cfg.MaxPull:
		return fail(CodeTooLarge, Summary{Restored: plan.LTXBytes, Cap: a.cfg.MaxPull})
	case !pull && plan.RestoredBytes > a.cfg.MaxRestored:
		return fail(CodeTooLarge, Summary{Restored: plan.RestoredBytes, Cap: a.cfg.MaxRestored})
	}
	need := plan.LTXBytes + plan.RestoredBytes
	if pull {
		need = plan.LTXBytes // the worker holds the files and restores nothing
	}
	if err := a.adm.grow(req.ID, need); err != nil {
		return fail(CodeNoRoom, Summary{})
	}

and replace everything from plan 1's stopPlanBound() after check B through its s := Summary{...} line with this, which chooses the stream, the chunk size, the bound and the summary, and keeps plan 1's two timing lines and its attemptStart and runStart uses (plan review finding 26):

	stopPlanBound() // the run is bounded by the worker and the owner's deadline
	stream, size, max := StreamRows, ChunkBytes, a.cfg.MaxOutput
	if pull {
		stream, size, max = StreamLTX, PullChunkBytes, plan.LTXBytes
	}
	// Plan 1's phase timings, kept for both ops (plan review finding 26;
	// Task 1 measures from them): a pull's first phase plans and has the
	// worker verify and hold the files, a query's plans and restores.
	held := "restored to"
	if pull {
		held = "held through"
	}
	a.logf("dbreads: %s: planned %d files (%d bytes) and %s %d in %s", req.ID, plan.Files, plan.LTXBytes, held, plan.TXID, time.Since(attemptStart).Round(time.Millisecond))
	runStart := time.Now()
	if err := writeJSONFrame(conn, frameRun, runMsg{SQL: req.SQL}); err != nil {
		return fail(CodeInterrupted, Summary{})
	}
	cw := newChunkWriter(ctx, a.cfg.Store, req.ID, pub, stream, a.cfg.ChunkEvery, max, size)
	endCode, sqlErr, rerr := relay(conn, cw)
	last, n, sum, werr := cw.close()
	a.logf("dbreads: %s: output %d bytes in %d confirmed chunks in %s", req.ID, n, last, time.Since(runStart).Round(time.Millisecond))
	s := Summary{TXID: plan.TXID, At: plan.At, Bytes: n, SHA256: sum, Error: sqlErr}
	if pull {
		s.Files = plan.Planned
		// A worker that sends more or less than it held is a check failure,
		// never an output cap.
		if werr == nil && rerr == nil && endCode == "" && (cw.capped() || n != plan.LTXBytes) {
			return a.ended(pub, req.ID, StateFailed, CodeCheckFailed, Summary{}, last), false
		}
	}

The existing switch on werr, cw.capped(), rerr and endCode follows unchanged. For a pull, cw.capped() is handled above, so its output-cap branch is unreachable. The worker's end for a pull is "" or check-failed, which the existing frame-end branch before check B already maps.

In internal/host/dbreads.go, add maxPull int64 to dbReadsEnv, parse CARLOS_DB_PULL_MAX_BYTES like CARLOS_DB_READ_MAX_BYTES (a whole number of bytes, else not armed), keep it only when it is below dbreads.MaxPullBytes, default dbreads.MaxPullBytes, and pass MaxPull: env.maxPull in armDBReads.

• [ ] Step 4: Run the tests

Run (sandbox off): go test ./internal/dbreads/ -count=1 -race && go test ./internal/host/ -run 'TestDBReads|TestPullCap' -count=1 Expected: PASS.

• [ ] Step 5: Add the pull to the root-runner tests

Now that the agent makes a pull, in internal/host/dbreads_rootrunner_linux_test.go add TestRootDBReadPullThroughTheInstalledUnits: as TestRootDBReadServesThroughTheInstalledUnits, but r.submit(dbreads.OpPull, "") (plan 1's submit already takes the op), and the answer is checked by opening every chunk with dbreads.StreamLTX, comparing the concatenation's SHA-256 with the summary's, and comparing it byte for byte with the planned files read from r.dir/replica/instances/shop.example.com in plan order. After it, privateTmpGone(t, 10*time.Second).

Build it (sandbox off): GOOS=linux GOARCH=arm64 go test -c -tags rootrunner -o "$TMPDIR/dbf3-rr.test" ./internal/host && rm "$TMPDIR/dbf3-rr.test". Expected: a clean build. Run TestRootDBReadPullThroughTheInstalledUnits on the root runner (plan 1 Task 15 Step 4's command) before the branch is reviewed, and record the run id.

• [ ] Step 6: Commit

git add internal/dbreads internal/host/dbreads.go internal/host/dbreads_test.go internal/host/dbreads_rootrunner_linux_test.go
git commit -m "dbreads: the agent's pull: its cap, its reservation, the ltx stream and the file list"

---

Task 6: The console forwards a pull

Waits for Task 2 (DBF-19, DBF-20 approved).

Files:

• Modify: internal/console/apidbforward.go (dbResidency's ForwardOps, admitDBRead's op check, dbForwardAuditDetail, two copy constants)
• Modify: internal/console/apidbforward_test.go
• Modify: approved_db_forward_copy_test.go (DBF-19, DBF-20)

Interfaces:

• Produces: ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}; dbForwardAuditDetail(op, region, sqlText string, at time.Time) string; dbfAuditPull, dbfAuditPullAt.

• [ ] Step 1: Write the failing tests

In internal/console/apidbforward_test.go, change TestListDoorMarksACapableRegion's expected ops to []string{dbreads.OpQuery, dbreads.OpPull}, change TestDBReadRouteRefusesBeforeAdmitting's "an op this console does not forward" case to op: "shell", add a case "a pull carrying SQL" (Op: pull, SQL: "SELECT 1", 400), and add:

func TestDBReadPullStreamsTheFiles(t *testing.T) {
	fx := newDBFFixture(t, true)
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	fx.startBox(t, dir)
	key, _ := dbreads.NewReplyKey()
	res := fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	lines := readLines(t, res.Body)
	res.Body.Close()
	last := lines[len(lines)-1].Result
	if last == nil || last.State != dbreads.StateDone {
		t.Fatalf("lines %+v", lines)
	}
	s, err := dbreads.OpenSummary(key, lines[0].Admitted, last.Sealed)
	if err != nil || len(s.Files) == 0 {
		t.Fatalf("summary %+v %v", s, err)
	}
	evs, _ := fx.st.Events(OpsAccountID, "shop", 10)
	want := fmt.Sprintf(dbfAuditPull, "uk")
	for _, e := range evs {
		if e.Action == "db-read" && e.Detail == want {
			return
		}
	}
	t.Fatalf("no %q audit row in %+v", want, evs)
}

• [ ] Step 2: Run them to verify they fail

Run: go test ./internal/console/ -run 'TestListDoorMarks|TestDBReadRouteRefuses|TestDBReadPull' -count=1 Expected: FAIL.

• [ ] Step 3: Make the changes

In apidbforward.go:

	dbfAuditPull   = "forwarded a database pull to region %s (cli)"          // DBF-19
	dbfAuditPullAt = "forwarded a database pull to region %s, as of %s (cli)" // DBF-20

dbResidency writes ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}. admitDBRead's op check becomes:

	switch body.Op {
	case dbreads.OpQuery:
		if err := dbquery.CheckSQL(body.SQL); err != nil || body.SQL == "" {
			msg := http.StatusText(http.StatusBadRequest)
			if err != nil {
				msg = err.Error()
			}
			jsonError(w, http.StatusBadRequest, msg)
			return none, "", false
		}
	case dbreads.OpPull:
		if body.SQL != "" {
			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
			return none, "", false
		}
	default:
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}

and dbForwardAuditDetail takes the op first:

func dbForwardAuditDetail(op, region, sqlText string, at time.Time) string {
	if op == dbreads.OpPull {
		if at.IsZero() {
			return fmt.Sprintf(dbfAuditPull, region)
		}
		return fmt.Sprintf(dbfAuditPullAt, region, at.UTC().Format(time.RFC3339))
	}
	sum := sha256.Sum256([]byte(sqlText))
	sha := hex.EncodeToString(sum[:])[:16]
	if at.IsZero() {
		return fmt.Sprintf(dbfAudit, region, len(sqlText), sha)
	}
	return fmt.Sprintf(dbfAuditAt, region, at.UTC().Format(time.RFC3339), len(sqlText), sha)
}

with its caller passing body.Op and TestDBReadQueuesASignedRequestAndAudits updated to pass dbreads.OpQuery. Add DBF-19 and DBF-20 to dbForwardCopyBuilt.

• [ ] Step 4: Run the tests

Run: go test ./internal/console/ -run 'TestDBRead|TestListDoor|TestDBForward' -count=1 && go test . -run 'TestDBForward' -count=1 Expected: PASS.

• [ ] Step 5: Commit

git add internal/console approved_db_forward_copy_test.go
git commit -m "console: forward a pull, and audit it (DBF-19, DBF-20)"

---

Task 7: The CLI's forwarded pull

Waits for Task 2 (DBF-17, DBF-18, DBF-24 approved).

Files:

• Modify: dbforward.go (the chunk sink, readDBForward generalised, the pull opener, dbPullForward, three constants)
• Modify: db.go (dbPull uses the forwarded pull when the marker offers it)
• Modify: dbforward_test.go
• Modify: approved_db_forward_copy_test.go (DBF-17, DBF-18, DBF-24)

Interfaces:

• Consumes: plan 2's readDBForward, openDBForward, errDBForward, newStream, fwdID; newConsoleClient, apiSession, apiAcct, dbTarget (main); dbdoor.ForwardBody; ltx.HeaderSize, ltx.TrailerSize, litestream.SnapshotLevel; dbdoor.RestoreLocal, dbdoor.Restored, lsfile.NewReplicaClient, litestream.LTXFilePath; copyOut, dbPullHolds, dbRestoreError, humanBytes.
• Produces: type dbChunkSink interface { add([]byte) error; count() int; chunk(i int) ([]byte, error) }; memSink, fileSink{dir string}; readDBForward(ctx, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, bound *dbfChunkBound) (string, *dbreads.Result, error) (the id and the result; openDBForward is called by the query path); openDBPull(ctx, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error); dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error); checkPullManifest(s dbreads.Summary) error; var dbPullSpoolWrite = os.WriteFile; dbfPullTooLarge, dbfFetching, dbfPullNoRoom; test helpers pullStream, pullStreamN, readPull, readPullIn, standInConsole, leftBehind.

• [ ] Step 1: Write the failing tests

Append to dbforward_test.go (crafted streams with real sealing: unit tests; Task 9 runs the binary):

// pullStream is a pull's body for the files of dir's replica, in one chunk.
func pullStream(t *testing.T, key *ecdh.PrivateKey, dir string, mutate func(s *dbreads.Summary, files [][]byte)) *stream {
	t.Helper()
	return pullStreamN(t, key, dir, 1, mutate)
}

// pullStreamN is pullStream with the files' bytes split across n chunks.
func pullStreamN(t *testing.T, key *ecdh.PrivateKey, dir string, n int, mutate func(s *dbreads.Summary, files [][]byte)) *stream {
	t.Helper()
	rd := lsfile.NewReplicaClient(dir)
	plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	var files [][]byte
	var all bytes.Buffer
	for _, f := range plan.Planned {
		rc, _ := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		b, _ := io.ReadAll(rc)
		rc.Close()
		files = append(files, b)
		all.Write(b)
	}
	sum := sha256.Sum256(all.Bytes())
	s := dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned, Bytes: int64(all.Len()), SHA256: hex.EncodeToString(sum[:])}
	if mutate != nil {
		mutate(&s, files)
	}
	st := newStream(t, key)
	b := all.Bytes()
	per := (len(b) + n - 1) / n
	for seq := 1; seq <= n; seq++ {
		part := b[min((seq-1)*per, len(b)):min(seq*per, len(b))]
		sealed, _ := dbreads.SealChunk(key.PublicKey(), fwdID, int64(seq), dbreads.StreamLTX, seq == n, part)
		st.line(dbreads.Line{Chunk: int64(seq), Data: sealed})
	}
	st.result(dbreads.StateDone, "", int64(n), s)
	return st
}

func readPull(t *testing.T, st *stream, key *ecdh.PrivateKey) (*dbdoor.Restored, error) {
	t.Helper()
	return readPullIn(t, st, key, t.TempDir())
}

func readPullIn(t *testing.T, st *stream, key *ecdh.PrivateKey, dir string) (*dbdoor.Restored, error) {
	t.Helper()
	sink := &fileSink{dir: filepath.Join(dir, "sealed")}
	id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes))
	if err != nil {
		return nil, err
	}
	return openDBPull(context.Background(), id, key, sink, *res, "shop.example.com", "uk", dir)
}

func TestForwardedPullRestoresTheFiles(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	key, _ := dbreads.NewReplyKey()
	r, err := readPull(t, pullStream(t, key, dir, nil), key)
	if err != nil {
		t.Fatal(err)
	}
	db, _ := sql.Open("sqlite", "file:"+r.Path+"?mode=ro")
	defer db.Close()
	var n int
	if err := db.QueryRow("SELECT count(*) FROM t").Scan(&n); err != nil || n != 30 {
		t.Fatalf("%d rows, %v", n, err)
	}
}

// Review Focus 3 and plan review findings 22 and 24: a file list that does
// not add up is DBF-13, never a panic, and no file is created before the
// list is checked whole. Mutations: drop checkPullManifest's per-file size
// bounds (red: "sizes -1 and N+1" panics or creates files); drop its sum
// (red: "sizes that do not sum" creates the first file before failing);
// drop the duplicate check (red: "a file twice" creates a file); drop the
// level bound (red: "level 12").
func TestForwardedPullRefusesFilesThatDoNotAddUp(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	for name, mutate := range map[string]func(s *dbreads.Summary, files [][]byte){
		"sizes that do not sum":      func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size++ },
		"a list that ends elsewhere": func(s *dbreads.Summary, _ [][]byte) { s.TXID++ },
		"no files":                   func(s *dbreads.Summary, _ [][]byte) { s.Files = nil },
		"sizes -1 and N+1": func(s *dbreads.Summary, _ [][]byte) {
			s.Files[1].Size += s.Files[0].Size + 1
			s.Files[0].Size = -1
		},
		"a file twice": func(s *dbreads.Summary, _ [][]byte) {
			s.Files[1] = s.Files[0]
			s.Bytes = 0
			for _, f := range s.Files {
				s.Bytes += f.Size
			}
		},
		"level 12":                  func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Level = 12 },
		"a range that runs back":    func(s *dbreads.Summary, _ [][]byte) { s.Files[0].MinTXID = s.Files[0].MaxTXID + 1 },
		"a size past the pull's cap": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size = dbreads.MaxPullBytes + 1 },
	} {
		key, _ := dbreads.NewReplyKey()
		out := t.TempDir()
		func() {
			defer func() {
				if p := recover(); p != nil {
					t.Errorf("%s: panic %v", name, p)
				}
			}()
			if _, err := readPullIn(t, pullStream(t, key, dir, mutate), key, out); err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") {
				t.Errorf("%s: %v", name, err)
			}
		}()
		if _, err := os.Stat(filepath.Join(out, "replica")); !os.IsNotExist(err) {
			t.Errorf("%s: a file was created before the list was checked (%v)", name, err)
		}
	}
}

// Plan review finding 37, at the spool: a chunk shorter than a sealed
// empty one creates no spool file, and a stream of tiny chunks stops at the
// bound's count, so the spool never holds more files than a read can make.
// (TestForwardedQueryBoundsChunkCountAndSize pins the count's derivation; a
// bound of 10 keeps this test small.) Mutation: add to the sink before
// admit (red: a file for the short chunk, eleven files for the tiny ones).
func TestForwardedPullSpoolsNoMalformedOrExcessChunk(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	for name, write := range map[string]func(s *stream) int{
		"a chunk shorter than its seal": func(s *stream) int {
			s.line(dbreads.Line{Chunk: 1, Data: []byte("short")})
			return 0
		},
		"tiny chunks past the count": func(s *stream) int {
			for seq := int64(1); seq <= 11; seq++ {
				sealed, _ := dbreads.SealChunk(key.PublicKey(), fwdID, seq, dbreads.StreamLTX, false, []byte("x"))
				s.line(dbreads.Line{Chunk: seq, Data: sealed})
			}
			return 10
		},
	} {
		s := newStream(t, key)
		want := write(s)
		dir := filepath.Join(t.TempDir(), "sealed")
		sink := &fileSink{dir: dir}
		_, _, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink,
			&dbfChunkBound{maxBytes: dbreads.MaxPullBytes, maxChunks: 10})
		files, _ := os.ReadDir(dir)
		if err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") || len(files) != want {
			t.Errorf("%s: %v, %d spool files, want %d", name, err, len(files), want)
		}
	}
}

// standInConsole answers the list door with the forward marker offering a
// pull, and the reads route with body (then holds the request open until
// the client goes when hold is set). It is a stand-in for the console, so
// that these tests run dbPull itself.
func standInConsole(t *testing.T, body []byte, hold bool) dbTarget {
	t.Helper()
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !strings.HasSuffix(r.URL.Path, "/reads") {
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusConflict)
			json.NewEncoder(w).Encode(dbdoor.ForwardBody{Error: "shop.example.com keeps its data in region uk.",
				Forward: 1, Region: "uk", ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}})
			return
		}
		io.Copy(io.Discard, r.Body)
		w.Header().Set("Content-Type", "application/x-ndjson")
		w.WriteHeader(http.StatusOK)
		w.Write(body)
		w.(http.Flusher).Flush()
		if hold {
			<-r.Context().Done()
		}
	}))
	t.Cleanup(ts.Close)
	return dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"},
		app: "shop", host: "shop.example.com"}
}

// leftBehind is what a pull left in its temp dir and at --out.
func leftBehind(t *testing.T, tmp, out string) {
	t.Helper()
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Errorf("something is at --out: %v", err)
	}
	if left, _ := filepath.Glob(filepath.Join(tmp, "carlos-db-*")); len(left) != 0 {
		t.Errorf("the pull's directory is still there: %v", left)
	}
}

// Review Focus 1 and plan review finding 23: a spool that fails part way,
// through dbPull itself, leaves nothing at --out and removes its directory.
// Not parallel: it sets TMPDIR and dbPullSpoolWrite. Mutation: drop
// dbPullForward's deferred RemoveAll (red: carlos-db-* is left).
func TestForwardedPullLeavesNothingWhenTheSpoolFails(t *testing.T) {
	tmp := t.TempDir()
	t.Setenv("TMPDIR", tmp)
	writes := 0
	dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error {
		if writes++; writes == 3 {
			return errors.New("no space left on device")
		}
		return os.WriteFile(name, b, mode)
	}
	t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile })
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	// The stand-in's chunks are sealed to a key the CLI does not hold; the
	// pull fails at the spool, before anything is opened.
	key, _ := dbreads.NewReplyKey()
	st := pullStreamN(t, key, dir, 4, nil)
	out := filepath.Join(t.TempDir(), "copy.db")
	if err := dbPull(context.Background(), standInConsole(t, st.buf.Bytes(), false), out); err == nil {
		t.Fatal("a spool that failed was accepted")
	}
	if writes != 3 {
		t.Fatalf("%d spool writes: the failure was not part way", writes)
	}
	leftBehind(t, tmp, out)
}

// Review Focus 4 and plan review finding 23: Ctrl-C while chunks are
// spooling, through dbPull itself, is "stopped", with nothing at --out and
// the directory removed. Not parallel: it sets TMPDIR and dbPullSpoolWrite.
// Mutation: drop dbPullForward's deferred RemoveAll (red). The binary's
// SIGINT is Task 9's TestDBForwardPullInterruptedLeavesNothing.
func TestForwardedPullCtrlCLeavesNothing(t *testing.T) {
	tmp := t.TempDir()
	t.Setenv("TMPDIR", tmp)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error {
		err := os.WriteFile(name, b, mode)
		cancel() // Ctrl-C once the first chunk is on disk
		return err
	}
	t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile })
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	key, _ := dbreads.NewReplyKey()
	st := pullStreamN(t, key, dir, 4, nil)
	// Admitted, picked and the first chunk only (newStream writes admitted
	// and picked first), then the stand-in holds the request.
	lines := bytes.SplitAfter(st.buf.Bytes(), []byte("\n"))
	out := filepath.Join(t.TempDir(), "copy.db")
	err := dbPull(ctx, standInConsole(t, bytes.Join(lines[:3], nil), true), out)
	if err == nil || err.Error() != "stopped" {
		t.Fatalf("%v", err)
	}
	leftBehind(t, tmp, out)
}

Update plan 2's query tests to the generalised signature: readFwd becomes

func readFwd(t *testing.T, s *stream, key *ecdh.PrivateKey) (dbForwardAnswer, string, error) {
	t.Helper()
	var status bytes.Buffer
	sink := &memSink{}
	id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes))
	if err != nil {
		return dbForwardAnswer{}, status.String(), err
	}
	ans, err := openDBForward(id, key, sink, *res, "shop.example.com", "uk")
	return ans, status.String(), err
}

and TestForwardedQueryCtrlCPrintsNothing and TestForwardedQueryBoundsChunkCountAndSize call readDBForward with dbfReading, &status, &memSink{}, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes) and check only its error. Add imports "database/sql", "errors", "io", "path/filepath", lsfile, ltx, dbdoortest (plan 2 already imports net/http, net/http/httptest and sync/atomic there). newStream (plan 2) writes admitted then picked, so lines[:3] in the Ctrl-C test is those and chunk 1; dbRestore turns the marker into errDBForward from the list door's first GET, whatever its path (plan 2 Task 8), which is why standInConsole answers the marker for every path but /reads.

• [ ] Step 2: Run them to verify they fail

Run: go test . -run 'TestForwardedPull|TestForwardedQuery' -count=1 Expected: FAIL: undefined names.

• [ ] Step 3: Write the code

In dbforward.go add the constants:

	dbfPullTooLarge = "%s's database is %s of replica files. carlos db pull can fetch at most %s from region %s."                    // DBF-17
	dbfFetching     = "fetching a copy of %s's database from region %s."                                                               // DBF-18
	dbfPullNoRoom   = "the server in region %s does not have room to prepare a copy of %s's database right now. try again later."    // DBF-24

the sinks:

// dbChunkSink holds a read's sealed chunks until its result arrives: in
// memory for a query (at most 64 MiB), in the pull's private directory for a
// pull (up to its cap).
type dbChunkSink interface {
	add(sealed []byte) error
	count() int
	chunk(i int) ([]byte, error) // 0-based
}

type memSink struct{ chunks [][]byte }

func (m *memSink) add(b []byte) error          { m.chunks = append(m.chunks, b); return nil }
func (m *memSink) count() int                  { return len(m.chunks) }
func (m *memSink) chunk(i int) ([]byte, error) { return m.chunks[i], nil }

// fileSink spools sealed chunks to dir, one file each, 0600 in a 0700
// directory the caller removes.
type fileSink struct {
	dir string
	n   int
}

func (f *fileSink) add(b []byte) error {
	if f.n == 0 {
		if err := os.MkdirAll(f.dir, 0o700); err != nil {
			return err
		}
	}
	f.n++
	return dbPullSpoolWrite(filepath.Join(f.dir, strconv.Itoa(f.n)), b, 0o600)
}

func (f *fileSink) count() int { return f.n }

func (f *fileSink) chunk(i int) ([]byte, error) {
	return os.ReadFile(filepath.Join(f.dir, strconv.Itoa(i+1)))
}

readDBForward takes the waiting sentence, the sink and plan 2's chunk bound (dbfChunkBound, plan review finding 37), and returns the id and the result instead of opening it:

func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, bound *dbfChunkBound) (string, *dbreads.Result, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	sc := bufio.NewScanner(body)
	sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
	var id, ended string
	for sc.Scan() {
		var l dbreads.Line
		if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
			return "", nil, unchecked
		}
		switch {
		case l.Admitted != "":
			if id != "" || !dbreads.ValidID(l.Admitted) {
				return "", nil, unchecked
			}
			id = l.Admitted
			fmt.Fprintln(status, fmt.Sprintf(reading, host, region))
		case l.Chunk != 0:
			// The bound is checked before the sink keeps anything, so a
			// malformed or excess chunk never reaches memory or the spool.
			if id == "" || l.Chunk != int64(sink.count())+1 || !bound.admit(l.Data) {
				return "", nil, unchecked
			}
			if err := sink.add(l.Data); err != nil {
				return "", nil, err
			}
		case l.Result != nil:
			if id == "" {
				return "", nil, unchecked
			}
			return id, l.Result, nil
		case l.Ended != "":
			ended = l.Ended
		}
	}
	if errors.Is(ctx.Err(), context.Canceled) {
		return "", nil, errors.New("stopped")
	}
	if ended == dbreads.EndedNotPicked {
		return "", nil, fmt.Errorf(dbfNotPicked, region)
	}
	return "", nil, fmt.Errorf(dbfUnfinished, host, region)
}

openDBForward reads its chunks from the sink: its signature becomes openDBForward(id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region string), with len(sealed) replaced by sink.count() and the loop reading sink.chunk(i) (an error from it is unchecked). dbQueryForward calls readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfReading, os.Stderr, &memSink{}, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)) and then openDBForward.

The pull's opener and verb:

// openDBPull opens a pull's result (spec §6, phase 2): a done one's chunks,
// opened in order, are the planned files concatenated; they are checked
// against the summary's count and hash and split by its file list into a
// litestream file replica, which today's restore and check turn into the
// copy. Any other state maps to its sentence.
func openDBPull(ctx context.Context, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	if res.Valid() != nil {
		return nil, unchecked
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		return nil, unchecked
	}
	if res.State != dbreads.StateDone {
		switch res.Code {
		case dbreads.CodeTooLarge:
			return nil, fmt.Errorf(dbfPullTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region)
		case dbreads.CodeNoRoom:
			return nil, fmt.Errorf(dbfPullNoRoom, region, host)
		}
		_, err := openDBForward(id, key, sink, res, host, region)
		return nil, err
	}
	if res.LastSeq != int64(sink.count()) || checkPullManifest(s) != nil {
		return nil, unchecked
	}
	root := filepath.Join(dir, "replica")
	h := sha256.New()
	fi, written := 0, int64(0)
	var out *os.File
	defer func() {
		if out != nil {
			out.Close()
		}
	}()
	for i := 0; i < sink.count(); i++ {
		sealed, err := sink.chunk(i)
		if err != nil {
			return nil, unchecked
		}
		seq := int64(i + 1)
		plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamLTX, seq == res.LastSeq, sealed)
		if err != nil {
			return nil, unchecked
		}
		h.Write(plain)
		for len(plain) > 0 {
			if fi >= len(s.Files) {
				return nil, unchecked
			}
			f := s.Files[fi]
			if out == nil {
				p := litestream.LTXFilePath(root, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID))
				if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
					return nil, err
				}
				if out, err = os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600); err != nil {
					return nil, err
				}
			}
			n := min(int64(len(plain)), f.Size-written)
			if _, err := out.Write(plain[:n]); err != nil {
				return nil, err
			}
			plain, written = plain[n:], written+n
			if written == f.Size {
				if err := out.Close(); err != nil {
					return nil, err
				}
				out, fi, written = nil, fi+1, 0
			}
		}
	}
	if fi != len(s.Files) || hex.EncodeToString(h.Sum(nil)) != s.SHA256 {
		return nil, unchecked
	}
	db := filepath.Join(dir, "replica.db")
	if err := dbdoor.RestoreLocal(ctx, lsfile.NewReplicaClient(root), db, s.TXID); err != nil {
		return nil, dbRestoreError(host, err)
	}
	os.RemoveAll(root)
	return &dbdoor.Restored{Path: db, Dir: dir, TXID: s.TXID, At: s.At}, nil
}

// checkPullManifest is the summary's file list checked whole before any
// file is created (plan review finding 22): every size a possible LTX file
// and within the cap, every level one litestream writes, every range a
// range, no file twice, sizes that sum without overflow to the summary's
// byte count, and a last file that ends at its transaction.
func checkPullManifest(s dbreads.Summary) error {
	bad := errors.New("the file list does not add up")
	if len(s.Files) == 0 || s.Bytes <= 0 || s.Bytes > dbreads.MaxPullBytes {
		return bad
	}
	type coord struct {
		level    int
		min, max uint64
	}
	seen := make(map[coord]bool, len(s.Files))
	var sum int64
	for _, f := range s.Files {
		c := coord{f.Level, f.MinTXID, f.MaxTXID}
		switch {
		case f.Size < ltx.HeaderSize+ltx.TrailerSize, f.Size > dbreads.MaxPullBytes,
			f.Level < 0, f.Level > litestream.SnapshotLevel,
			f.MinTXID < 1, f.MaxTXID < f.MinTXID, seen[c],
			sum > dbreads.MaxPullBytes-f.Size: // sum + f.Size would pass the cap
			return bad
		}
		seen[c] = true
		sum += f.Size
	}
	if sum != s.Bytes || s.Files[len(s.Files)-1].MaxTXID != s.TXID {
		return bad
	}
	return nil
}

// dbPullSpoolWrite writes one spooled chunk: os.WriteFile, a seam so a test
// can fail the spool part way (plan review finding 23).
var dbPullSpoolWrite = os.WriteFile

// dbPullForward is a pull through the forwarded read. The spool, the files
// and the copy live in one private 0700 directory under the system temp
// dir, removed on every path but success, where the caller removes it after
// copyOut, as it does today's restore directory.
func dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error) {
	ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound)
	defer cancel()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		return nil, err
	}
	at := ""
	if !t.at.IsZero() {
		at = t.at.UTC().Format(time.RFC3339Nano)
	}
	body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpPull, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	if err != nil {
		return nil, err
	}
	res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"),
		bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}})
	if err != nil {
		if errors.Is(ctx.Err(), context.Canceled) {
			return nil, errors.New("stopped")
		}
		return nil, fmt.Errorf(dbfUnfinished, t.host, fwd.Region)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
		return nil, apiErrorFromBody(res.StatusCode, raw)
	}
	dir, err := os.MkdirTemp("", "carlos-db-")
	if err != nil {
		return nil, err
	}
	if err := os.Chmod(dir, 0o700); err != nil {
		os.RemoveAll(dir)
		return nil, err
	}
	done := false
	defer func() {
		if !done {
			os.RemoveAll(dir)
		}
	}()
	sink := &fileSink{dir: filepath.Join(dir, "sealed")}
	id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfFetching, os.Stderr, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes))
	if err != nil {
		return nil, err
	}
	r, err := openDBPull(ctx, id, key, sink, *result, t.host, fwd.Region, dir)
	if err != nil {
		return nil, err
	}
	os.RemoveAll(sink.dir)
	done = true
	return r, nil
}

In db.go, dbPull becomes:

func dbPull(ctx context.Context, t dbTarget, out string) error {
	r, err := dbRestore(ctx, t)
	var fwd *errDBForward
	if errors.As(err, &fwd) {
		if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpPull) {
			// A console that forwards queries and not pulls (plan 2's).
			return fmt.Errorf(dbfNoPull, t.host, fwd.door.Forward.Region)
		}
		r, err = dbPullForward(ctx, t, *fwd.door.Forward)
	}
	if err != nil {
		return err
	}
	defer os.RemoveAll(r.Dir)
	if err := copyOut(ctx, r.Path, out); err != nil {
		return err
	}
	fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out)
	fmt.Fprint(stdout, dbPullHolds(r, t.at))
	fmt.Fprintf(stdout, "It is a copy. Nothing you do to it reaches %s.\n", t.host)
	return nil
}

Add strconv, litestream, lsfile and ltx imports to dbforward.go. Add DBF-17, DBF-18 and DBF-24 to dbForwardCopyBuilt.

• [ ] Step 4: Run the tests

Run: go test . -run 'TestForwardedPull|TestForwardedQuery|TestDBForward|TestDBPull|TestCopyOut' -count=1 && go vet . && GOOS=windows GOARCH=amd64 go build . && GOOS=darwin GOARCH=arm64 go build . Expected: PASS and clean builds.

• [ ] Step 5: Commit

git add dbforward.go dbforward_test.go db.go approved_db_forward_copy_test.go
git commit -m "carlos db pull: the forwarded pull, spooled, checked, restored locally (DBF-17, DBF-18, DBF-24)"

---

Task 8: The CLI page (DBF-23)

Waits for Task 2.

Files:

• Modify: docs/site/cli.md (the two paragraphs DBF-30 wrote)
• Modify: approved_db_forward_copy_test.go (DBF-23 in, DBF-30 out of dbForwardCopyBuilt)

• [ ] Step 1: Swap the ids and run the gate

In dbForwardCopyBuilt, replace "DBF-30" with "DBF-23". Run: go test . -run 'TestDBForwardBuiltCopyIsApplied' -count=1. Expected: FAIL: the page still carries DBF-30.

• [ ] Step 2: Replace the paragraphs

Replace DBF-30's two one-line paragraphs in the carlos db section of docs/site/cli.md with DBF-23's approved two paragraphs, one line each, a blank line between them.

• [ ] Step 3: Run the gate and the docs check

Run: go test . -run 'TestDBForward' -count=1 && go test ./internal/docsite/ -count=1 Expected: PASS.

• [ ] Step 4: Commit

git add docs/site/cli.md approved_db_forward_copy_test.go
git commit -m "docs: carlos db pull reads another region's instances (DBF-23)"

---

Task 9: Acceptance: the pull through the real CLI, MinIO, and size

Files:

• Modify: dbforward_e2e_test.go
• Modify: internal/console/dbforward_minio_test.go

Interfaces:

• Consumes: plan 2's dbfE2E, newDBFE2E, run, betweenTheSyncs, scriptBox, e2eScript, TestDBForwardMinio's helpers.

• [ ] Step 1: Write the tests

Append to dbforward_e2e_test.go:

func sha256File(t *testing.T, path string) string {
	t.Helper()
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	sum := sha256.Sum256(b)
	return hex.EncodeToString(sum[:])
}

// The pull's parity: a forwarded pull and an in-region pull of the same
// replica save byte-identical copies, for the newest copy and as of a time,
// and print the same lines but for the host.
func TestDBForwardPullParityThroughTheCLI(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true})
	for _, at := range []string{"", e.betweenTheSyncs(t)} {
		dir := t.TempDir()
		args := []string{"db", "pull", "--app", "shop"}
		if at != "" {
			args = append(args, "--at", at)
		}
		here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere, "--out", filepath.Join(dir, "here.db"))...)
		if err != nil {
			t.Fatalf("in region: %v\n%s", err, hereErr)
		}
		far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar, "--out", filepath.Join(dir, "far.db"))...)
		if err != nil {
			t.Fatalf("forwarded: %v\n%s", err, farErr)
		}
		if sha256File(t, filepath.Join(dir, "here.db")) != sha256File(t, filepath.Join(dir, "far.db")) {
			t.Fatalf("at %q the copies differ", at)
		}
		if strings.ReplaceAll(far, dbfE2EFar, dbfE2EHere) != strings.ReplaceAll(here, filepath.Join(dir, "here.db"), filepath.Join(dir, "far.db")) {
			t.Fatalf("at %q: %q vs %q", at, here, far)
		}
		if !strings.Contains(farErr, fmt.Sprintf(dbfFetching, dbfE2EFar, "uk")) {
			t.Fatalf("no DBF-18: %s", farErr)
		}
	}
}

// T11 for a pull: a hash mismatch saves nothing at --out. Everything else
// is a pull the CLI would accept: far's real planned files, whole and
// valid, a manifest that passes checkPullManifest, and a summary whose only
// fault is its SHA-256 (plan review finding 32). Mutation: openDBPull skips
// the hash comparison (red: the files restore and a copy is saved).
func TestDBForwardPullSavesNothingForABadAnswer(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	stop := e.scriptBox(t, func(b *e2eScript) {
		ctx := context.Background()
		rd := lsfile.NewReplicaClient(filepath.Join(e.root, "instances", dbfE2EFar))
		plan, err := dbdoor.PlanFrom(ctx, rd, time.Time{})
		if err != nil {
			t.Errorf("plan far's replica: %v", err)
			return
		}
		var all bytes.Buffer
		for _, f := range plan.Planned {
			rc, err := rd.OpenLTXFile(ctx, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
			if err != nil {
				t.Errorf("open %+v: %v", f, err)
				return
			}
			io.Copy(&all, rc)
			rc.Close()
		}
		b.picked(time.Now().Add(time.Minute))
		sealed, _ := dbreads.SealChunk(b.pub, b.id, 1, dbreads.StreamLTX, true, all.Bytes())
		b.store.Put(ctx, dbreads.ChunkKey(b.id, 1), sealed, "")
		wrong := sha256.Sum256([]byte("not these bytes"))
		summary, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned,
			Bytes: int64(all.Len()), SHA256: hex.EncodeToString(wrong[:])})
		r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: 1, Box: "uk-1", Sealed: summary})
		b.store.Put(ctx, dbreads.ResultKey(b.id), r, "")
	})
	defer stop()
	out := filepath.Join(t.TempDir(), "far.db")
	_, stderr, err := e.run(t, "db", "pull", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	if err == nil || !strings.Contains(stderr, "did not check out") {
		t.Fatalf("%q %v", stderr, err)
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
}

// Review Focus 4 and plan review finding 23, through the binary: SIGINT
// while the CLI is spooling a pull prints "stopped", saves nothing at --out
// and removes the pull's directory. The scripted box (a stand-in) writes two
// chunks and never a result, so the CLI is mid-pull when the signal lands.
// Mutation: drop dbPullForward's deferred RemoveAll (red: carlos-db-* is
// left in the CLI's TMPDIR).
func TestDBForwardPullInterruptedLeavesNothing(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	stop := e.scriptBox(t, func(b *e2eScript) {
		b.picked(time.Now().Add(time.Minute))
		for seq := int64(1); seq <= 2; seq++ {
			sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamLTX, false, bytes.Repeat([]byte{'x'}, 4096))
			b.store.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "")
		}
	})
	defer stop()
	spool := t.TempDir()
	out := filepath.Join(t.TempDir(), "far.db")
	cmd := exec.Command(e.bin, "db", "pull", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	// The last TMPDIR wins (os/exec keeps the last of a duplicated key).
	cmd.Env = append(carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...), "TMPDIR="+spool)
	cmd.Dir = t.TempDir()
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	deadline := time.Now().Add(30 * time.Second)
	for {
		if got, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*", "sealed", "2")); len(got) == 1 {
			break
		}
		if time.Now().After(deadline) {
			cmd.Process.Kill()
			cmd.Wait()
			t.Fatalf("the CLI never spooled two chunks: %s", stderr.String())
		}
		time.Sleep(50 * time.Millisecond)
	}
	if err := cmd.Process.Signal(os.Interrupt); err != nil {
		t.Fatal(err)
	}
	if err := cmd.Wait(); err == nil || !strings.Contains(stderr.String(), "stopped") {
		t.Fatalf("%v: %s", err, stderr.String())
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
	if left, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*")); len(left) != 0 {
		t.Fatalf("the pull's directory is still there: %v", left)
	}
}

// A console from plan 2 forwards queries and not pulls. Its marker is
// played here by a wrapper that drops "pull" from forward_ops on the way
// out: the new CLI says DBF-8 and saves nothing.
func TestDBForwardPullOnAQueryOnlyConsole(t *testing.T) {
	queryOnly := func(t *testing.T, h http.Handler) (string, []string) {
		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			rec := httptest.NewRecorder()
			h.ServeHTTP(rec, r)
			body := rec.Body.Bytes()
			if rec.Code == http.StatusConflict {
				body = bytes.Replace(body, []byte(`"forward_ops":["query","pull"]`), []byte(`"forward_ops":["query"]`), 1)
			}
			for k, vs := range rec.Header() {
				w.Header()[k] = vs
			}
			w.Header().Del("Content-Length")
			w.WriteHeader(rec.Code)
			w.Write(body)
		}))
		t.Cleanup(ts.Close)
		return ts.URL, nil
	}
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: queryOnly})
	out := filepath.Join(t.TempDir(), "far.db")
	_, stderr, err := e.run(t, "db", "pull", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfNoPull, dbfE2EFar, "uk")) {
		t.Fatalf("%q %v", stderr, err)
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
}

The wrapper buffers each response, so it is used only where no response streams (the list door's 409 is all this test reads).

In internal/console/dbforward_minio_test.go, right after T1's query loop, add a pull of shop through MinIO:

	// A pull through MinIO: the planned files, sealed as the ltx stream,
	// adding up to the summary.
	pkey, _ := dbreads.NewReplyKey()
	pres := dfx.openRead(t, shop, member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(pkey.PublicKey())})
	plines := readLines(t, pres.Body)
	pres.Body.Close()
	presult := plines[len(plines)-1].Result
	if presult == nil || presult.State != dbreads.StateDone {
		t.Fatalf("pull: %+v", plines[len(plines)-1])
	}
	psum, err := dbreads.OpenSummary(pkey, plines[0].Admitted, presult.Sealed)
	if err != nil || len(psum.Files) == 0 || psum.TXID != psum.Files[len(psum.Files)-1].MaxTXID {
		t.Fatalf("pull summary %+v %v", psum, err)
	}
	h := sha256.New()
	for _, l := range plines {
		if l.Chunk == 0 {
			continue
		}
		plain, err := dbreads.OpenChunk(pkey, plines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == presult.LastSeq, l.Data)
		if err != nil {
			t.Fatalf("pull chunk %d: %v", l.Chunk, err)
		}
		h.Write(plain)
	}
	if hex.EncodeToString(h.Sum(nil)) != psum.SHA256 {
		t.Fatal("the pulled bytes do not add up to the summary")
	}

with "crypto/sha256" and "encoding/hex" added to that file's imports.

• [ ] Step 2: Run them

Add lsfile "github.com/benbjohnson/litestream/file" and "github.com/superfly/ltx" to the file's imports (plan 2's harness already imports os/exec, crypto/sha256, encoding/hex and dbdoor). The far instance's replica is the litestream file layout under e.root/instances/far.example.com, which betweenTheSyncs already reads.

Run (sandbox off): go test . -run 'TestDBForwardPull' -count=1 -v Expected: PASS. The MinIO subtest runs in CI's MinIO step.

• [ ] Step 3: The size run (only on Paul's say-so)

The spec's long-request concern for a pull is a full-size one. Ask Paul before running it: it writes a database of the cap's size (Task 1's number, up to 2 GiB) and its replica, about three times that on disk. With his go, on dev (not DOCS-T1): an instance in dev's uk region whose database is near the cap (seed it by a carlos run of a script that inserts randomblob rows until the database is the size wanted, then let it sleep so its replica syncs), then time carlos db pull --app <app> --instance <host> --out "$TMPDIR/dbf-big.db" from a laptop through the dev console. Record the time, the CLI's peak spool size, and that it finished inside 22 minutes; remove the file and the instance after. Report on the branch's discussion.

• [ ] Step 4: Commit

git add dbforward_e2e_test.go internal/console/dbforward_minio_test.go
git commit -m "db forward: the pull's parity, bad answer and MinIO tests"

---

Task 10: Mutation checks, census, weights, merge, CI, roll

Files:

• Create: internal/provision/testdata/plan14-root-reads-db-region-forward-3.tsv
• Modify: internal/provision/plan14_root_reads_test.go
• Create: docs/superpowers/notes/census-proposal-db-region-forward-3.md
• Modify: .amadan/test-weights.tsv

• [ ] Step 1: Mutation checks

In a tree copy under $TMPDIR, as plans 1 and 2 did; every row must turn red:

| Mutation | Test that must fail | |---|---| | holdAndSend skips verifyLTX | TestWorkerHoldRefusesACorruptFile | | the agent's pull sends run before check B | TestPullTooLargeAndNoChunkBeforeCheckB (the recorded frames hold a run) | | openDBPull skips the hash comparison | TestDBForwardPullSavesNothingForABadAnswer (real files and a valid manifest; only the digest is wrong: plan review finding 32) | | readDBForward adds to the sink before bound.admit (plan review finding 37) | TestForwardedPullSpoolsNoMalformedOrExcessChunk | | the agent's pull drops its two timing lines (plan review finding 26) | TestPullShipsThePlannedFiles | | checkPullManifest drops the sum | TestForwardedPullRefusesFilesThatDoNotAddUp ("sizes that do not sum": a file is created before the failure) | | checkPullManifest drops the per-file size bounds | TestForwardedPullRefusesFilesThatDoNotAddUp ("sizes -1 and N+1": a panic, or a file created) | | checkPullManifest drops the duplicate or the level check | TestForwardedPullRefusesFilesThatDoNotAddUp ("a file twice", "level 12") | | NewAgent keeps a MaxPull above MaxPullBytes | TestNewAgentClampsMaxPull (plan review finding 24: TestPullCapOnlyLowers reads only dbReadsSettings, which clamps on its own) | | dbReadsSettings keeps a CARLOS_DB_PULL_MAX_BYTES above the cap | TestPullCapOnlyLowers | | dbPullForward drops its deferred RemoveAll | TestForwardedPullLeavesNothingWhenTheSpoolFails, TestForwardedPullCtrlCLeavesNothing, TestDBForwardPullInterruptedLeavesNothing |

• [ ] Step 2: Merge main, census, 9B8

Merge main; run plan 1 Task 17's dump with the prefixes dbforward.go, db.go, internal/dbreads/, internal/console/apidbforward, internal/host/dbreads; write plan14-root-reads-db-region-forward-3.tsv (rev3=PROPOSED-db-region-forward-3; each @legacy fingerprint computed by plan 1's TestZZDBForwardFingerprints from the row's own fields) with plans 1 and 2's basis templates: worker units as the db-read worker at J4 ("verifies and holds the planned files in its private scratch and sends them back after the agent's check B"), agent and console units re-frozen at their plan 1 and plan 2 dispositions with the pull appended, CLI units as the carlos CLI at J4 ("spools sealed chunks to a 0700 directory under the system temp dir, writes the opened files as a litestream file replica there, restores and checks them with dbdoor.RestoreLocal, and removes the directory on every path"); envelopes re-pinned. Load it last with the previous seal checked first; set the new seal. Run TestPlan9B8RecordWriteCensus; re-pin any changed entry at its class in a dbRegionForward3WriteCensusInventory layered on plan 2's. Write the census proposal and post it on #280.

• [ ] Step 3: Test weights

Measure the new root and console tests with targeted -json runs (plan 2 Task 14 Step 6) and add rows for those of a second or more.

• [ ] Step 4: Push, describe, CI, review, merge

As plan 2 Task 14 Steps 7 and 8: final local checks, push, the branch description (what plan 3 adds; Task 1's measurement and Paul's confirmation of the cap; the mutation table; the root-runner run id of TestRootDBReadPullThroughTheInstalledUnits; whether the size run was done and its numbers; the census proposal; the copy state), green CI on the tip, the whole-branch review, merge main again, squash, merge behind git merge-base --is-ancestor origin/main HEAD.

• [ ] Step 5: The roll, each live step on Paul's go

Dev boxes, dev console, check (a forwarded carlos db pull and carlos db pull --at of a dev uk instance from a laptop: both save a copy whose PRAGMA integrity_check is ok, with today's three lines on stdout and DBF-18 on stderr), then the flagship boxes and console. The lifecycle rule is already in place from plan 2. The per-region bucket that keeps a pull's ciphertext in the data's region is plan 5 (spec §14, ruling 1), not this plan.