# `carlos db` region forward, plan 2: console and CLI for `query` and `query --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 query` and `carlos db query --at` against an instance whose data lives in another region: the console admits the read, signs and queues it for plan 1's boxes, holds one synchronous HTTP request open while it relays the box's sealed answer, and the CLI buffers, opens and checks the answer before printing it; plus the docs, the copy, and the deployment bucket's `db-reads/` lifecycle rule with the approval plan for applying it.

**Architecture:** The console's db door keeps every in-region behaviour byte for byte. A cross-region record on a console with `CARLOS_DB_FORWARD=1` and a root key gets an additive 409 marker on the list door (DBF-14 to an old CLI). A new route, `POST /api/cli/apps/{account}/{app}/db/{host}/reads`, runs the db door's whole gate once, checks capability and liveness on complete evidence only, takes admission counts held by the request itself, writes the audit row, then the signed request and its queue entry, bumps the serial on the request's own context (a request that has ended writes only `cancel`), and then streams newline-delimited JSON (admitted, picked, chunks, wait lines, result) inside the same request, reading the box's `result.json` before listing its chunks. Nothing about a read outlives its request except one best-effort `cancel`. The CLI switches to that route on the marker, makes an ephemeral key, and prints nothing until a done result's summary checks out.

**Tech Stack:** Go 1.26, `net/http` (`http.ResponseController` read and write deadlines and `Flush`), `bufio.Scanner` over NDJSON, plan 1's `internal/dbreads` and `internal/dbreads/dbreadstest`, MinIO in `.amadan/ci.d/30-minio-contract`, OpenTofu module tests.

**Spec:** `docs/superpowers/specs/2026-10-22-db-region-forward-design.md` (rev 6: rev 5, accepted by Astra in round 5 and approved by Paul for building on 2026-10-22 with the rulings in §16, plus the four clarifications the plan review forced, §5.4, §5.5, §5.7 and §9.3). Read §5.2 to §5.7, §7.1, §9, §10, §11 and §15 before starting. Plan 1 (`2026-10-22-db-region-forward-plan-1-box.md`) must have merged: this plan consumes its packages by the names its tasks list.

**Branch:** `db-region-forward-2-query`, cut from amadan `main` after plan 1 merged.

## Global Constraints

Every task's requirements include this section.

- **Production-wired acceptance.** Acceptance goes through the real console server (`console.NewServer` served by `console.NewHTTPServer`), the real CLI binary (`buildCarlos`, env through `carlosTestEnv`, `cmd.Dir` a scratch directory), the real edge (`edge.New`), real stores (MinIO in CI for T1 and T16) and the real verify and seal code. The box in console and CLI tests is plan 1's real agent and worker code joined in process by `dbreadstest`, which is not acceptance for the units; plan 1's root-runner tests are. A scripted box (a test that writes `picked`, chunks and `result.json` itself) is a stand-in and says so.
- **Targeted local runs; CI runs the full suite.** Run only the touched packages with `-run`, for example `go test ./internal/console/ -run 'TestDBRead' -count=1`. Never `make test`, never the whole console or root package. Build, `go vet`, `gofmt -l` and the cross-compiles run locally. CI's full run on the tip is the merge gate.
- **Cross-compile gate.** `GOOS=darwin GOARCH=arm64 go build ./...` and `GOOS=windows GOARCH=amd64 go build ./...` before every push. The CLI's forwarded path is portable; nothing here is Linux-only.
- **Copy needs review before it ships, with no em dashes.** No DBF row reaches code or docs before Paul approves it (Task 1). The record is `docs/superpowers/plans/2026-10-22-db-region-forward-copy.md`; the gate is `approved_db_forward_copy_test.go`. Lower case after a full stop. The CLI and console's existing sentences that capitalise are left as they are (ruling 4). Internal fragments and log lines are not rows.
- **No tenant names.** `shop`, `shop.example.com`, `other.example.com`, `member@example.com`, regions `ie` and `uk`.
- **Go cache.** Leave `GOCACHE`, `GOMODCACHE` and `GOTMPDIR` at their defaults; run the go command with the sandbox off if the cache is read-only. Tests that open sockets or build the CLI binary run with the sandbox off.
- **Census overlays per branch, loaded LAST.** This branch adds `internal/provision/testdata/plan14-root-reads-db-region-forward-2.tsv`, loaded in `TestPlan14RootReadCensus` after whatever main loads last at the time of the merge (plan 1's `db-region-forward-1` at the earliest; re-read the test after every merge of main), with the previous seal checked first. Rows carry `rev3=PROPOSED-db-region-forward-2`; the ruling is asked of the census owner on #280.
- **Merge main first,** before asking for review and again immediately before merging, rebuilding the census on the merged tree; merge only with `git merge-base --is-ancestor origin/main HEAD && amadan branch merge ...`.
- **Test weights.** Every new top-level test in `internal/console` or the root package that takes a second or more gets a row in `.amadan/test-weights.tsv` (Task 13).
- **No CI-only scripts under `scripts/`.** The MinIO setup for T1 and T16 is added to `.amadan/ci.d/30-minio-contract`, which already owns the MinIO server.
- **New host tests call `t.Parallel()`** unless they touch env, cwd, umask or package vars. (This plan adds none.)
- **Scratch** goes under `$TMPDIR` or `t.TempDir()`, removed by name.
- **Wire shapes are additive.** The list door's 409 keeps its `{"error"}` sentence; the marker fields are additions. A console with `CARLOS_DB_FORWARD` unset answers byte-identically to main.
- **Production infrastructure needs scoped approval.** The lifecycle rule is prepared and planned here; it is applied only after Paul approves the written plan (AGENTS.md), dev first.
- **Spec numbers, verbatim.** The request's pre-header work bounded by 20 s; `admitted` flushed with the headers; a progress line at least every 15 s; result read about once a second, before the chunk listing; picked awaited for 2 minutes after `issued_at` (DBF-3); the owner's deadline plus 1 minute (DBF-12), and never more than 22 minutes after `issued_at`; a write stalled for 30 s ends the request; the CLI's context is 23 minutes; open reads per person 2, per console 16 (429 with `Retry-After`); the request budget is the db door's 600 per token per minute; output at most 64 MiB.

## 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. **Ctrl-C mid-stream.** A person interrupts while chunks are arriving. They expect `stopped`, no rows on stdout, and the console to write `cancel` and free their slot. Test: Task 8, `TestForwardedQueryCtrlCPrintsNothing`, and Task 7, `TestDBReadAdmissionBounds`.
2. **A full-size chunk line.** A 256 KiB chunk sealed and base64-encoded is a line of about 350 KiB, past `bufio.Scanner`'s 64 KiB default. A person expects a large result to arrive. Test: Task 8, `TestForwardedQueryReadsAFullSizeChunkLine`.
3. **A `picked` whose deadline is absurd.** Any box can write `picked` (spec §5.6), so its deadline may be far in the future. A person expects the console to give up 22 minutes after `issued_at` at the latest. Test: Task 7, `TestDBReadRequestBoundClampsAForgedDeadline`.
4. **A `result.json` that is not a result.** Garbage, or a state outside the vocabulary. A person expects the console to keep waiting until its deadline and the CLI to say DBF-12, never a panic or a relayed garbage line. Test: Task 7, `TestDBReadIgnoresAMalformedResult`.
5. **A home-region record seen from a regional console.** Its record region is empty. A person expects today's residency sentence, never a forward marker naming region "". Test: Task 5, `TestListDoorKeepsMainsSentenceForTheHomeRegion`.

## Plan review, round 1

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

- **2** (`hookStore` collides with `internal/console/temphook_test.go:16`): renamed `dbReadHookStore`.
- **6** (fixtures create over an existing object): `putFixtureObject` reads the etag and replaces with it.
- **7** (Task 6's admission test needs Task 7's stream): the test is Task 7's, `TestDBReadAdmissionBounds`.
- **11** (writes after the request ended): the audit row is written first, before any box can see the read; after the queue entry, a request that has ended writes `cancel` and returns, with no serial bump; the bump runs on the request's context (`announceDBRead`). Spec §9.3 says so in rev 6. Partly declined, below.
- **12** (bounds do not hold blocking work): phase deadline contexts, every store call through `dbfCall` (5 s, wait lines while it blocks), the bound checked before the result line, write deadlines clamped to the bound plus 10 s. Tests: `TestDBReadStalledStoreKeepsTheStreamAndItsBound`, `TestDBReadSlowRelayEndsAtTheBound`.
- **13** (a slow body outlives the pre-header bound): the read deadline is `preHeader` until admission, then the read's bound. Test: `TestDBReadStalledBodyIsCutAtThePreHeaderBound`.
- **18** (T15): both of a person's slots freed before and after pickup, the seventeenth read across nine people, and a restart under open reads.
- **19** (T16 passes without late output): a gate holds the real box at publication until the console's request has ended, by disconnect and by the bound.
- **20** (result-before-list): its own operation-order check and its own mutation row.
- **21** (the console half): the stream's last log line, `relayed N chunks (B bytes) in T, ended <why>`.

### Declined plan findings

- **11, in part: making the audit insert itself cancellation-aware.** `RecordActivityContext` takes a context, but its claim plumbing ends in `claimRawRecordActivity`'s context-free `Exec`, shared by every audited route (`internal/console/store.go:2588`, `2668`). Changing that is a change to the claim SQL layer and its census, for one local SQLite insert. The order makes it unnecessary: the row is written while the request is open (checked just before) and before any object a box can see, so a request that ends during the insert leaves the row of an admission and no read. The serial bump, the one bucket write, is request-bound.

## Plan review, round 2

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

- **25** (`dbStream` is `internal/console/apidb.go:475`'s function): the relay type is `dbReadRelay`. Every new identifier in the three plans was then checked against `origin/main`; this was the only collision.
- **28** (admission could still publish after cancellation): the request's context is checked immediately before each of admission's four writes (audit row, `request.json`, queue entry, serial), with `cancel` once anything is published. Test: `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel`, one case per step, through the `s.dbf.beforeWrite` seam. Not closed by the checks alone, and not closable without a transactional store: a request that ends between a check and its write still sees that write land; the next check then writes `cancel`, and a box that has already elected stops at its next `cancel` read (every 5 s). The audit insert's own cancellation stays declined (round 1).
- **30** (pickup windows shorter than `issued_at`'s truncation): every accelerated window leaves the observation it needs after up to a second of truncation; production's timestamps are unchanged.
- **31** (the completeness guard unwitnessed): `TestDBReadResultWaitsForAChunkTheListingMissed`.
- **33** (the serial mutation never ran): "ended during the serial read" cancels inside the bump, with a store that refuses a cancelled write as S3 does.
- **34** (T7 lacked the real CLI and console through backhaul): Task 11 runs both edges in package `edge`'s test binary as a helper process and the real CLI and console in the root package; `checkProgressAtTheCLI` watches the CLI's stderr and socket, through the local edge as well.
- **35** (the result-line deadline check has a witness): `TestDBReadResultAfterTheDeadlineIsNotRelayed`, with a `ResponseWriter` that holds the last chunk's line.
- **36** (SQL JSON would rewrite): the CLI refuses it with DBF-32 before sending, the console refuses U+FFFD with 400. Tests at the console, the CLI and the binary.
- **37** (unbounded chunk count): `dbfChunkBound` caps the count and refuses a short chunk, or anything after an empty one, before it is kept.
- **38** (the extension's mutation stayed green): `TestDBReadSetsBothReadDeadlines` watches the connection's read deadlines.

None declined beyond round 1's audit-insert item.

## Census impact

- **Plan 14 root-read census:** one overlay, `plan14-root-reads-db-region-forward-2.tsv`, `rev3=PROPOSED-db-region-forward-2`:
  - new accepted-legacy console units (identity: the console, `cmd/console serve` under `console.service`, user `carlos-console`): the reads route's handler, admission, stream and line writer, the residency answer, the evidence rule, `NewHTTPServer`; changed legacy console units re-frozen at main's disposition (`dbGate`, `Server.registerCLIRoutes` if it reads, `cmd/console` `serve` for the forwarding switch and `NewHTTPServer`);
  - new accepted-legacy CLI units (identity: the carlos CLI on a person's machine, as the person): the forwarded query, its line reader and opener; `dbdoor` `doorError` re-frozen for the additive marker;
  - `infra/modules/deployment-buckets/main.tf` re-frozen as an external unit at main's disposition with the two `db-reads/` lifecycle rules appended;
  - envelopes re-pinned: `.`, `cmd/console`, `internal/console`, `internal/dbdoor`, `internal/dbreads`, `infra/modules/deployment-buckets`. The seal moves.
- **9B8 record-write census:** `Server.registerCLIRoutes` changes as a `key-site` for the one route; new `nonrecord` `Put` sites for the request, the queue entry and `cancel`, all conditional creates. No `Delete` site, no record writer.
- **Claim routes:** `route POST /api/cli/apps/{account}/{app}/db/{host}/reads` joins `internal/console/testdata/claim-routes.txt`.
- **go.mod:** unchanged.

## Copy strings

Reviewed in Task 1 and built by the tasks named:

| id | file | text | task |
|---|---|---|---|
| DBF-1 | internal/console/apidbforward.go | <host> keeps its data in region <region>. no server in that region can read databases yet, so carlos db cannot read it for now. | 5 |
| DBF-2 | internal/console/apidbforward.go | <host> keeps its data in region <region>. no server in that region is answering right now. try again in a few minutes. | 5 |
| DBF-3 | dbforward.go | no server in region <region> picked up the read within 60 seconds. try again in a few minutes. | 8 |
| DBF-4 | dbforward.go | <host>'s database is <size> once restored. carlos db can query at most <cap> in region <region>. | 8 |
| DBF-5 | dbforward.go | the server in region <region> does not have room to restore <host>'s database right now. try again later. | 8 |
| DBF-6 | dbforward.go | the result was cut off at 64 MB. narrow the query with WHERE or LIMIT and run it again. | 8 |
| DBF-7 | dbforward.go | the query ran for 5 minutes and was stopped. narrow it and run it again. | 8 |
| DBF-8 | dbforward.go | <host> keeps its data in region <region>. carlos db pull cannot fetch a copy from there yet. carlos db query can read it. | 8 |
| DBF-9 | internal/console/apidbforward.go | too many database reads are open. wait a minute and try again. | 6 |
| DBF-10 | internal/console/apidbforward.go | <host> is moving between regions. read its database once the move has finished. | 6 |
| DBF-11 | dbforward.go | reading a copy of <host>'s database in region <region>. | 8 |
| DBF-12 | dbforward.go | the read of <host>'s database in region <region> did not finish, so nothing was printed. try again. | 8 |
| DBF-13 | dbforward.go | the answer from region <region> did not check out, so carlos db showed none of it. try again. | 8 |
| DBF-14 | internal/console/apidbforward.go | <host> keeps its data in region <region>. this version of carlos cannot read it from there. run carlos update, then try again. | 5 |
| DBF-15 | dbforward.go | <host> changed while the read was waiting. try again. | 8 |
| DBF-16 | dbforward.go | <host> is moving between regions. read its database once the move has finished. | 8 |
| DBF-21 | internal/console/apidbforward.go | forwarded a database query to region <region>. the query was <n> bytes, sha256 <sha> (cli) | 6 |
| DBF-22 | internal/console/apidbforward.go | forwarded a database query to region <region>, as of <time>. the query was <n> bytes, sha256 <sha> (cli) | 6 |
| DBF-32 | dbforward.go | this query cannot be sent to region <region>, because it holds bytes that are not UTF-8 text or the character U+FFFD. fix the query and run it again. | 8 |
| DBF-30 | docs/site/cli.md | (two paragraphs; spec §10) | 9 |

Unchanged and reused (spec §10): `dbQueryStatus`, the `dbRestoreError` sentences, SQLite's own text for `sql-error`, `stopped`, the `checkQuerySQL` refusals, the residency sentence when forwarding is off.

## File structure

| File | Responsibility |
|---|---|
| `internal/dbreads/lines.go` (create) | The reads route's wire: `Line`, `ReadBody`, wait and ended words, `MaxLineBytes`. |
| `internal/dbdoor/forward.go` (create), `client.go` (modify) | The list door's additive marker: `ForwardBody` written by the console, `DoorError.Forward` read by the CLI. |
| `internal/console/httpserver.go` (create) | `NewHTTPServer`: the console's `http.Server`, shared by `cmd/console` and T7. |
| `internal/console/apidbforward.go` (create) | The forwarding switch and signer, the evidence rule, the residency answer, the reads route (admission and stream), the DBF console copy. |
| `internal/console/apidb.go` (modify) | `dbGate`'s residency case calls the residency answer. |
| `internal/console/server.go` (modify) | `Config.DBForward`; the read limits and timing seams. |
| `internal/console/authcli.go` (modify), `testdata/claim-routes.txt` (modify) | The route. |
| `cmd/console/main.go` (modify) | `CARLOS_DB_FORWARD`; `NewHTTPServer`. |
| `dbforward.go` (create), `db.go` (modify) | The CLI's forwarded query and the DBF CLI copy. |
| `docs/site/cli.md` (modify) | DBF-30. |
| `infra/modules/deployment-buckets/main.tf` (modify), `tests/db_reads.tftest.hcl` (create) | The lifecycle rule (J4), T17. |
| `docs/superpowers/notes/db-reads-lifecycle-approval.md` (create) | The scoped approval plan for applying it. |
| `.amadan/ci.d/30-minio-contract` (modify) | Buckets, users and the PASS check for T1 and T16. |
| Tests: `internal/console/apidbforward_test.go`, `internal/console/dbforward_minio_test.go`, `dbforward_test.go`, `dbforward_e2e_test.go`, `internal/edge/dbforward_backhaul_helper_test.go` (create) | T1, T6, T7, T8, T11, T14, T15, T16, T18. |

---

### Task 1: Copy review for the console, CLI and docs rows

**Files:**
- Modify: `docs/superpowers/plans/2026-10-22-db-region-forward-copy.md`

**Interfaces:**
- Consumes: the record plan 1 created; `dbForwardCopyIDs` and `dbForwardCopyBuilt` in `approved_db_forward_copy_test.go`.

- [ ] **Step 1: Run the review**

Invoke the `copy-review` skill with rows DBF-1 to DBF-16, DBF-21, DBF-22, DBF-30 and DBF-32, each with its context from spec §10 (where it appears and when), neutral example values (`shop.example.com`, region `uk`, `2.0GB`, `64 MB`). Read `result.json` before relaunching the review server: a relaunch has destroyed a submitted review before.

- [ ] **Step 2: Apply Paul's edits verbatim**

Copy each approved text into the record exactly, set its status to `approved`, and run `go test . -run 'TestDBForwardCopy' -count=1` (inventory, no em dash). Expected: PASS. If Paul rewrote a row so that a later task's format string must change, the task uses the record's text, not this plan's.

- [ ] **Step 3: Commit**

```bash
git add docs/superpowers/plans/2026-10-22-db-region-forward-copy.md
git commit -m "db forward: plan 2's copy approved"
```

---

### Task 2: The reads route's wire and the list door's marker

**Files:**
- Create: `internal/dbreads/lines.go`, `internal/dbreads/lines_test.go`
- Create: `internal/dbdoor/forward.go`, `internal/dbdoor/forward_test.go`
- Modify: `internal/dbdoor/client.go:44-66` (`DoorError` gains `Forward`) and `client.go:169-187` (`doorError` reads it)

**Interfaces:**
- Produces: `dbreads.Line{Admitted, Picked string; Chunk int64; Data []byte; Wait, Ended string; Result *Result}`; `dbreads.ReadBody{Op, SQL, At, ReplyKey string}`; `dbreads.WaitPickup`, `WaitReading`, `EndedNotPicked`, `EndedNoResult`; `dbreads.MaxLineBytes`; `dbdoor.ForwardBody{Error string; Forward int; Region string; ForwardOps []string}`; `dbdoor.Forward{Region string; Ops []string}`; `DoorError.Forward *Forward`.

- [ ] **Step 1: Write the failing tests**

Create `internal/dbreads/lines_test.go`:

```go
package dbreads

import (
	"bytes"
	"encoding/json"
	"testing"
)

// Each line carries exactly one outcome, and a full-size sealed chunk fits
// MaxLineBytes once base64-encoded.
func TestLinesAreOneOutcomeEach(t *testing.T) {
	t.Parallel()
	for _, l := range []Line{
		{Admitted: "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"},
		{Picked: "uk-1"},
		{Chunk: 1, Data: []byte("sealed")},
		{Wait: WaitPickup},
		{Ended: EndedNotPicked},
		{Result: &Result{State: StateDone, LastSeq: 1}},
	} {
		b, err := json.Marshal(l)
		if err != nil {
			t.Fatal(err)
		}
		var m map[string]any
		json.Unmarshal(b, &m)
		keys := len(m)
		if l.Chunk != 0 {
			keys-- // a chunk line is two fields, chunk and data
		}
		if keys != 1 {
			t.Fatalf("%s carries %d outcomes", b, keys)
		}
	}
	full, _ := json.Marshal(Line{Chunk: 1, Data: bytes.Repeat([]byte{0xff}, ChunkBytes+SealOverhead)})
	if len(full)+1 > MaxLineBytes {
		t.Fatalf("a full chunk line is %d bytes, over MaxLineBytes", len(full))
	}
}
```

Create `internal/dbdoor/forward_test.go`:

```go
package dbdoor

import (
	"encoding/json"
	"io"
	"net/http"
	"strings"
	"testing"
)

func doorResponse(status int, body string) *http.Response {
	return &http.Response{StatusCode: status, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(body))}
}

// The marker is additive: the sentence stays what every CLI prints, and only
// a 409 with forward 1 and a region carries a Forward.
func TestDoorErrorReadsTheForwardMarker(t *testing.T) {
	t.Parallel()
	body, _ := json.Marshal(ForwardBody{Error: "shop.example.com keeps its data in region uk.", Forward: 1, Region: "uk", ForwardOps: []string{"query"}})
	var door *DoorError
	err := doorError(doorResponse(http.StatusConflict, string(body)))
	door = err.(*DoorError)
	if door.Message != "shop.example.com keeps its data in region uk." || door.Forward == nil ||
		door.Forward.Region != "uk" || len(door.Forward.Ops) != 1 || door.Forward.Ops[0] != "query" {
		t.Fatalf("door = %+v %+v", door, door.Forward)
	}
	for name, raw := range map[string]string{
		"a plain refusal":   `{"error":"no"}`,
		"forward 0":         `{"error":"no","forward":0,"region":"uk"}`,
		"no region":         `{"error":"no","forward":1}`,
		"a region that is not one": `{"error":"no","forward":1,"region":"../x"}`,
	} {
		if d := doorError(doorResponse(http.StatusConflict, raw)).(*DoorError); d.Forward != nil {
			t.Errorf("%s: a marker", name)
		}
	}
	if d := doorError(doorResponse(http.StatusNotFound, string(body))).(*DoorError); d.Forward != nil {
		t.Error("a 404 carried a marker")
	}
}
```

- [ ] **Step 2: Run them to verify they fail**

Run: `go test ./internal/dbreads/ -run 'TestLines' -count=1; go test ./internal/dbdoor/ -run 'TestDoorErrorReadsTheForwardMarker' -count=1`
Expected: FAIL, undefined names.

- [ ] **Step 3: Write `lines.go`**

```go
package dbreads

// The reads route's 200 body is newline-delimited JSON, one Line per line
// (spec §5.5 step 6). MaxLineBytes bounds one line: a ChunkBytes chunk,
// sealed and base64-encoded, is about 350 KiB.
const MaxLineBytes = 1 << 20

// Wait lines name what the console is waiting on; ended lines say why the
// response ends without a result.
const (
	WaitPickup     = "pickup"
	WaitReading    = "reading"
	EndedNotPicked = "not-picked" // the CLI says DBF-3
	EndedNoResult  = "no-result"  // the CLI says DBF-12
)

// Line is one outcome: admitted (the read's id), picked (the owner box),
// one chunk in sequence, a wait, an ended, or the result, which comes last.
type Line struct {
	Admitted string  `json:"admitted,omitempty"`
	Picked   string  `json:"picked,omitempty"`
	Chunk    int64   `json:"chunk,omitempty"`
	Data     []byte  `json:"data,omitempty"`
	Wait     string  `json:"wait,omitempty"`
	Ended    string  `json:"ended,omitempty"`
	Result   *Result `json:"result,omitempty"`
}

// ReadBody is the POST …/db/{host}/reads body (spec §5.5 step 2).
type ReadBody struct {
	Op       string `json:"op"`
	SQL      string `json:"sql"`
	At       string `json:"at"`
	ReplyKey string `json:"reply_key"`
}
```

The `ended` line is an additive outcome line (spec §5.5: "from here every outcome is a line in that body"). It lets the CLI tell "nobody picked the read" (DBF-3) from "the read did not finish" (DBF-12) when a console restart or a dropped connection also ends a response with no result.

- [ ] **Step 4: Write `forward.go` and teach `doorError` the marker**

Create `internal/dbdoor/forward.go`:

```go
package dbdoor

import "regexp"

// ForwardBody is the db door's 409 for an instance a forwarding console can
// read in its own region (db region forward spec §5.5 step 1). It is
// additive to the {"error"} body every CLI reads, so an old CLI prints Error
// as it is (DBF-14) and a new one switches to the forwarded read.
type ForwardBody struct {
	Error      string   `json:"error"`
	Forward    int      `json:"forward"`
	Region     string   `json:"region"`
	ForwardOps []string `json:"forward_ops"`
}

// Forward is a DoorError's marker: where the data is, and which operations
// the console forwards there.
type Forward struct {
	Region string
	Ops    []string
}

var forwardRegion = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}$`)
```

In `client.go`, add the field to `DoorError`:

```go
	// Forward is the list door's forward marker on a 409 (db region
	// forward spec §5.5): the instance is readable through the forwarded
	// read in Forward.Region. nil for every other refusal.
	Forward *Forward
```

and in `doorError`, decode the marker fields beside `error`:

```go
	var body struct {
		Error      string   `json:"error"`
		Forward    int      `json:"forward"`
		Region     string   `json:"region"`
		ForwardOps []string `json:"forward_ops"`
	}
	e := &DoorError{Status: res.StatusCode}
	if json.Unmarshal(raw, &body) == nil {
		e.Message = strings.TrimSpace(body.Error)
		if res.StatusCode == http.StatusConflict && body.Forward == 1 && forwardRegion.MatchString(body.Region) {
			e.Forward = &Forward{Region: body.Region, Ops: body.ForwardOps}
		}
	}
```

- [ ] **Step 5: Run the tests**

Run: `go test ./internal/dbreads/ -run 'TestLines' -count=1 && go test ./internal/dbdoor/ -count=1`
Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add internal/dbreads/lines.go internal/dbreads/lines_test.go internal/dbdoor
git commit -m "db forward: the reads route's lines and the list door's additive marker"
```

---

### Task 3: The console's switch, signer, limits and `http.Server`

**Files:**
- Create: `internal/console/httpserver.go`
- Create: `internal/console/apidbforward.go` (this task's part: the switch, the signer, the timings)
- Create: `internal/console/apidbforward_test.go`
- Modify: `internal/console/server.go:566` (`Config.DBForward`), `server.go:920-923` (fields) and `server.go:1460` (initialisation)
- Modify: `cmd/console/main.go:384-390` (the config), `main.go:586-590` (the server), and add `dbForward()`

**Interfaces:**
- Consumes: `release.RootSigner`, `newDBStreamLimits` (`apidb.go`), `dbreads.OpenPerConsole`, `dbreads.OpenPerPerson`.
- Produces: `console.NewHTTPServer(h http.Handler) *http.Server`; `Config.DBForward bool`; `(*Server).dbForwardSigner() (release.Signer, bool)`; `Server.dbReads *dbStreamLimits`; `Server.dbf dbForwardTimes` with fields `poll, wait, stall, preHeader, pickup, grace, bound, call time.Duration` and the test seam `beforeWrite func(step string)`.

- [ ] **Step 1: Write the failing tests**

Create `internal/console/apidbforward_test.go` (later tasks add to it):

```go
package console

import (
	"crypto/ed25519"
	"crypto/rand"
	"testing"
	"time"
)

// The console forwards only with the switch on, a root key whose public
// half is the deployment's, and no host claims (spec §5.5, §9.1, §11.1).
func TestDBForwardSignerNeedsTheSwitchAndTheRootKey(t *testing.T) {
	t.Parallel()
	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
	_, other, _ := ed25519.GenerateKey(rand.Reader)
	for name, c := range map[string]struct {
		cfg  Config
		want bool
	}{
		"on, root key":       {Config{DBForward: true, Key: priv, ReleasePub: pub}, true},
		"off":                {Config{Key: priv, ReleasePub: pub}, false},
		"no key":             {Config{DBForward: true, ReleasePub: pub}, false},
		"not the root's key": {Config{DBForward: true, Key: other, ReleasePub: pub}, false},
		"host claims":        {Config{DBForward: true, Key: priv, ReleasePub: pub, HostClaim: true}, false},
	} {
		s := &Server{cfg: c.cfg}
		if _, ok := s.dbForwardSigner(); ok != c.want {
			t.Errorf("%s: %v", name, ok)
		}
	}
}

func TestNewHTTPServerKeepsTheConsolesTimeouts(t *testing.T) {
	t.Parallel()
	srv := NewHTTPServer(nil)
	if srv.ReadHeaderTimeout != 10*time.Second || srv.ReadTimeout != 5*time.Minute || srv.WriteTimeout != 0 {
		t.Fatalf("%+v", srv)
	}
}
```

- [ ] **Step 2: Run them to verify they fail**

Run: `go test ./internal/console/ -run 'TestDBForwardSigner|TestNewHTTPServer' -count=1`
Expected: FAIL, undefined names.

- [ ] **Step 3: Write the code**

Create `internal/console/httpserver.go`:

```go
package console

import (
	"net/http"
	"time"
)

// NewHTTPServer is the console's http.Server (cmd/console serves with it):
// ReadHeaderTimeout 10 s, ReadTimeout 5 minutes for request bodies, and no
// WriteTimeout, because the doors that stream (the db door, the forwarded
// read) set their own per-write deadlines. A door that holds a request
// longer than ReadTimeout extends its own read deadline (the import door,
// the forwarded read); the forwarded read's long-request tests (T7) serve
// through this constructor.
func NewHTTPServer(h http.Handler) *http.Server {
	return &http.Server{
		Handler:           h,
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       5 * time.Minute,
	}
}
```

In `cmd/console/main.go` replace the inline `&http.Server{...}` at lines 586 to 590 with `httpSrv := console.NewHTTPServer(srv)`, keeping the comment above it. Add, beside `secretsEnvelopeV2`:

```go
// dbForward parses CARLOS_DB_FORWARD: "1" turns on forwarding a carlos db
// read to the region that holds the data (db region forward spec §11.1).
// Unset, every db door answer is byte-identical to main. Anything else stops
// the boot, as for the other switches.
func dbForward() bool {
	switch v := strings.TrimSpace(os.Getenv("CARLOS_DB_FORWARD")); v {
	case "":
		return false
	case "1":
		return true
	default:
		log.Fatalf("console: CARLOS_DB_FORWARD %q is not 1 (or unset)", v)
		return false
	}
}
```

and set `DBForward: dbForward(),` in the `console.Config` literal beside `Region: placement`.

In `internal/console/server.go`, add to `Config` after `Region`:

```go
	// DBForward is CARLOS_DB_FORWARD=1: a carlos db read of an instance in
	// another region is forwarded to a capable box there (db region
	// forward spec). It also needs Key to be the deployment root's.
	DBForward bool
```

and to `Server` beside `dbStreams`:

```go
	// dbReads bounds open forwarded reads per person and per console, held
	// by each read's request (spec §5.7, J5); dbf is its timing.
	dbReads *dbStreamLimits
	dbf     dbForwardTimes
```

and in `NewServer` beside the `dbStreams` line:

```go
	s.dbReads = newDBStreamLimits(dbreads.OpenPerConsole, dbreads.OpenPerPerson)
	s.dbf = defaultDBForwardTimes()
```

Create `internal/console/apidbforward.go`:

```go
// The forwarded `carlos db` read (db region forward spec
// docs/superpowers/specs/2026-10-22-db-region-forward-design.md): the list
// door's marker for a cross-region instance, and the one route that admits a
// read, queues it for a box in the data's region, and relays the box's sealed
// answer inside the same request.
package console

import (
	"crypto/ed25519"
	"time"

	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/release"
)

// dbForwardTimes is the forwarded read's timing (spec §5.2, §5.5). Tests
// shrink it; production uses defaultDBForwardTimes.
type dbForwardTimes struct {
	poll      time.Duration // how often the console reads picked, result and chunks
	wait      time.Duration // a progress line at least this often
	stall     time.Duration // a write or flush that blocks this long ends the request
	preHeader time.Duration // everything before the 200 (backhaul waits 30 s for headers)
	pickup    time.Duration // picked must appear this long after issued_at
	grace     time.Duration // after the owner's deadline
	bound     time.Duration // the request's own bound after issued_at
	call      time.Duration // one store call inside the stream (plan review finding 12)

	// beforeWrite, nil in production, runs just before admission checks the
	// request's context ahead of each write ("audit", "request", "queue",
	// "serial"): a test ends the request there (plan review finding 28).
	beforeWrite func(step string)
}

func defaultDBForwardTimes() dbForwardTimes {
	return dbForwardTimes{
		poll: time.Second, wait: 15 * time.Second, stall: 30 * time.Second, preHeader: 20 * time.Second,
		pickup: dbreads.ConsoleWaitsPicked, grace: dbreads.OwnerGrace, bound: dbreads.RequestBound,
		call: 5 * time.Second,
	}
}

// dbForwardSigner is the root signer when this console forwards: the switch
// on, no host claims (a host-claim console answers with the residency
// sentence, as it refuses carlos run), and a key whose public half is the
// deployment root a box verifies against. A console holding only a scoped
// key cannot sign a request, and so cannot forward (spec §5.5).
func (s *Server) dbForwardSigner() (release.Signer, bool) {
	if !s.cfg.DBForward || s.cfg.HostClaim || s.cfg.Key == nil {
		return release.Signer{}, false
	}
	pub, ok := s.cfg.Key.Public().(ed25519.PublicKey)
	if !ok || !pub.Equal(s.cfg.ReleasePub) {
		return release.Signer{}, false
	}
	return release.RootSigner(s.cfg.Key), true
}
```

- [ ] **Step 4: Run the tests and build the console**

Run: `go test ./internal/console/ -run 'TestDBForwardSigner|TestNewHTTPServer' -count=1 && go build ./cmd/console/`
Expected: PASS and a clean build.

- [ ] **Step 5: Commit**

```bash
git add internal/console/httpserver.go internal/console/apidbforward.go internal/console/apidbforward_test.go internal/console/server.go cmd/console/main.go
git commit -m "console: CARLOS_DB_FORWARD, the root signer check, the read limits, NewHTTPServer"
```

---

### Task 4: The evidence rule (T14)

**Files:**
- Modify: `internal/console/apidbforward.go` (add the rule)
- Modify: `internal/console/apidbforward_test.go` (T14)

**Interfaces:**
- Consumes: `edgesSnapshot` (`edgesfile.Snapshot`), `s.liveEdges(ctx)` (`regioncandidates.go:203`, which is `edgesfile.ReadLiveSnapshot`), `dbreads.Capability`.
- Produces: `dbForwardEvidence(snap edgesSnapshot, err error, region string) string`; constants `dbForwardNone`, `dbForwardDead`; test helpers `putEdgesFile(t, s release.Store, box, placement string, caps []string, draining bool)` and `putBoxStatus(t, s release.Store, box string, at time.Time)`.

- [ ] **Step 1: Write the failing test**

Append to `internal/console/apidbforward_test.go`:

```go
// putFixtureObject creates key, or replaces it under its current etag: a
// fixture may republish a box's files, as a box does every pass. Protocol
// writes stay create-only; this helper is for fixtures only (plan review
// finding 6: DirStore.Put refuses a create over an existing key).
func putFixtureObject(t *testing.T, s release.Store, key string, body []byte) {
	t.Helper()
	etag := ""
	if _, cur, err := s.Get(context.Background(), key); err == nil {
		etag = cur
	}
	if _, err := s.Put(context.Background(), key, body, etag); err != nil {
		t.Fatal(err)
	}
}

// putEdgesFile writes control/edges/<box>.json as a box publishes it.
func putEdgesFile(t *testing.T, s release.Store, box, placement string, caps []string, draining bool) {
	t.Helper()
	body, _ := json.Marshal(map[string]any{"v": 1, "box": box, "eip": "203.0.113.10", "placement": placement,
		"capabilities": caps, "draining": draining, "hosts": map[string]any{}})
	putFixtureObject(t, s, "control/edges/"+box+".json", body)
}

// putBoxStatus writes status/boxes/<box>.json generated at at.
func putBoxStatus(t *testing.T, s release.Store, box string, at time.Time) {
	t.Helper()
	body, _ := json.Marshal(map[string]any{"generated_at": at.UTC(), "label": box})
	putFixtureObject(t, s, "status/boxes/"+box+".json", body)
}

// T14 (codex 10): only complete evidence, both flags, refuses up front.
// Incomplete evidence of any kind admits the read and lets the pickup
// window decide.
func TestDBForwardEvidenceRefusesOnlyOnCompleteEvidence(t *testing.T) {
	t.Parallel()
	now := time.Now()
	capable := []string{dbreads.Capability}
	for name, c := range map[string]struct {
		setup func(s release.Store)
		want  string
	}{
		"a capable live box": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, false)
			putBoxStatus(t, s, "uk-1", now)
		}, ""},
		"no capable box": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			putBoxStatus(t, s, "uk-1", now)
		}, dbForwardNone},
		"a capable box that is draining": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, true)
			putBoxStatus(t, s, "uk-1", now)
		}, dbForwardNone},
		"capable but stale": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, false)
			putBoxStatus(t, s, "uk-1", now.Add(-time.Hour))
		}, dbForwardDead},
		"capable in another region only": {func(s release.Store) {
			putEdgesFile(t, s, "ie-1", "ie", capable, false)
			putBoxStatus(t, s, "ie-1", now)
		}, dbForwardNone},
		"a status snapshot missing": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
		}, ""},
		"an edges file unreadable": {func(s release.Store) {
			s.Put(context.Background(), "control/edges/uk-1.json", []byte("{not json"), "")
			putBoxStatus(t, s, "uk-1", now)
		}, ""},
		"a fresh snapshot with no edges file": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			putBoxStatus(t, s, "uk-1", now)
			putBoxStatus(t, s, "uk-2", now)
		}, ""},
		"status snapshots unreadable": {func(s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			s.Put(context.Background(), "status/boxes/uk-1.json", []byte("{not json"), "")
		}, ""},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			s := release.NewDirStore(t.TempDir())
			c.setup(s)
			snap, err := readLiveEdgesSnapshot(context.Background(), s, now)
			if got := dbForwardEvidence(snap, err, "uk"); got != c.want {
				t.Fatalf("evidence %q, want %q (complete %v, boxes %v)", got, c.want, snap.Complete, snap.BoxesComplete)
			}
		})
	}
}
```

Add `"context"`, `"encoding/json"`, the `dbreads` and `release` imports to the test file.

- [ ] **Step 2: Run it to verify it fails**

Run: `go test ./internal/console/ -run 'TestDBForwardEvidence' -count=1`
Expected: FAIL, `dbForwardEvidence` undefined.

- [ ] **Step 3: Write the rule**

Append to `internal/console/apidbforward.go` (add `"slices"`):

```go
const (
	dbForwardNone = "none" // DBF-1: no capable, non-draining box publishes in the region
	dbForwardDead = "dead" // DBF-2: capable boxes, none live
)

// dbForwardEvidence is spec §7.1's rule. It reads Complete and
// BoxesComplete, never the error alone: ReadLiveSnapshot answers a nil
// error when status snapshots cannot be loaded and reports the gap in those
// flags (codex 10). Incomplete evidence comes first and wins over both
// refusals: the request is written and the pickup window decides.
func dbForwardEvidence(snap edgesSnapshot, err error, region string) string {
	if err != nil || !snap.Complete || !snap.BoxesComplete {
		return ""
	}
	capable, live := 0, 0
	for _, f := range snap.Files {
		if f.Placement != region || f.Draining || !slices.Contains(f.Capabilities, dbreads.Capability) {
			continue
		}
		capable++
		if f.Fresh {
			live++
		}
	}
	switch {
	case capable == 0:
		return dbForwardNone
	case live == 0:
		return dbForwardDead
	}
	return ""
}
```

- [ ] **Step 4: Run it**

Run: `go test ./internal/console/ -run 'TestDBForwardEvidence' -count=1 -v`
Expected: PASS, nine subtests. If "a fresh snapshot with no edges file" answers `none`, `ReadLiveSnapshot` did not mark the evidence incomplete: read its ruling (A) comment and fix the fixture, never the rule.

- [ ] **Step 5: Commit**

```bash
git add internal/console/apidbforward.go internal/console/apidbforward_test.go
git commit -m "console: the forwarded read's evidence rule (T14)"
```

---

### Task 5: The list door's answer for a cross-region instance

Waits for Task 1 (DBF-1, DBF-2, DBF-14 approved).

**Files:**
- Modify: `internal/console/apidbforward.go` (the residency answer and three copy constants)
- Modify: `internal/console/apidb.go:186-188` (`dbGate`'s `ErrResidency` case)
- Modify: `internal/console/apidbforward_test.go`
- Modify: `approved_db_forward_copy_test.go` (`dbForwardCopyBuilt` gains DBF-1, DBF-2, DBF-14)

**Interfaces:**
- Consumes: `dbForwardSigner`, `dbForwardEvidence` (Tasks 3, 4), `dbdoor.ForwardBody` (Task 2), `s.dbResidencyRefusal`, `registry.FleetPlacement`, `dbreads.ValidRegion`.
- Produces: `(*Server).dbResidency(w http.ResponseWriter, r *http.Request, rec provision.Record)`; `dbfNoCapable`, `dbfNoneLive`, `dbfUpdateCarlos`; the test fixture `dbfFixture` and `newDBFFixture(t, forward bool) *dbfFixture`.

- [ ] **Step 1: Write the failing tests**

Append to `internal/console/apidbforward_test.go`:

```go
const dbfHost = "shop.example.com"

// dbfFixture is a console in ie with forwarding on or off, a member, and
// shop.example.com's record in uk with a capable, live uk box published.
type dbfFixture struct {
	*consoleFixture
	member string
}

func newDBFFixture(t *testing.T, forward bool) *dbfFixture {
	t.Helper()
	fx := newConsoleFixture(t)
	fx.srv.cfg.DBForward = forward
	fx.srv.cfg.Region = "ie"
	fx.srv.cfg.HibernateBucket = "data-ie"
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "member@example.com", "code-dbf")
	putShopRecord(t, fx, provision.Record{Host: dbfHost, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	putEdgesFile(t, fx.rel, "uk-1", "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-1", time.Now())
	return &dbfFixture{consoleFixture: fx, member: member}
}

func (fx *dbfFixture) listURL() string {
	return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + dbfHost + "/ltx/9"
}

// T8, the console's half: forwarding off is main's answer byte for byte.
func TestListDoorWithForwardingOffIsMains(t *testing.T) {
	fx := newDBFFixture(t, false)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	want := `{"error":"shop.example.com keeps its data in region uk. This console can only read data kept in region ie."}`
	if status != http.StatusConflict || strings.TrimSpace(body) != want {
		t.Fatalf("%d %s", status, body)
	}
}

func TestListDoorMarksACapableRegion(t *testing.T) {
	fx := newDBFFixture(t, true)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	var got dbdoor.ForwardBody
	json.Unmarshal([]byte(body), &got)
	if status != http.StatusConflict || got.Forward != 1 || got.Region != "uk" ||
		!slices.Equal(got.ForwardOps, []string{dbreads.OpQuery}) ||
		got.Error != fmt.Sprintf(dbfUpdateCarlos, dbfHost, "uk") {
		t.Fatalf("%d %s", status, body)
	}
}

func TestListDoorRefusesOnCompleteEvidence(t *testing.T) {
	fx := newDBFFixture(t, true)
	putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) || strings.Contains(body, `"forward"`) {
		t.Fatalf("no capable box: %d %s", status, body)
	}
	putEdgesFile(t, fx.rel, "uk-2", "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-2", time.Now().Add(-time.Hour))
	status, _, body = rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoneLive, dbfHost, "uk")) {
		t.Fatalf("none live: %d %s", status, body)
	}
}

// Review Focus 5: a home-region record seen from a regional console keeps
// main's sentence; there is no region to forward to by name.
func TestListDoorKeepsMainsSentenceForTheHomeRegion(t *testing.T) {
	fx := newDBFFixture(t, true)
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "home.example.com", CreatedAt: time.Now().UTC()})
	u := fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/home.example.com/ltx/9"
	status, _, body := rawReq(t, http.MethodGet, u, fx.member, "")
	if status != http.StatusConflict || strings.Contains(body, `"forward"`) || !strings.Contains(body, "This console can only read data kept in") {
		t.Fatalf("%d %s", status, body)
	}
}
```

`TestListDoorWithForwardingOffIsMains`'s `want` is main's `dbResidencyRefusal` for this fixture (`dbRegionName` names a non-empty region `region <code>`). Run it on main's code first (`git stash` the branch's console changes, or run it in Step 2 before Step 3) and confirm main produces exactly these bytes; that is what pins "byte-identical to main". `putShopRecord` creates; `TestListDoorKeepsMainsSentenceForTheHomeRegion` creates a second record.

- [ ] **Step 2: Run them to verify they fail**

Run: `go test ./internal/console/ -run 'TestListDoor' -count=1`
Expected: `TestListDoorWithForwardingOffIsMains` and `TestListDoorKeepsMainsSentenceForTheHomeRegion` PASS already; the other two FAIL.

- [ ] **Step 3: Write the answer**

Append to `internal/console/apidbforward.go` (imports `fmt`, `net/http`, `dbdoor`, `provision`, `registry`):

```go
// The console's forwarded-read sentences (copy record DBF-1, DBF-2, DBF-14).
const (
	dbfNoCapable    = "%s keeps its data in region %s. no server in that region can read databases yet, so carlos db cannot read it for now."
	dbfNoneLive     = "%s keeps its data in region %s. no server in that region is answering right now. try again in a few minutes."
	dbfUpdateCarlos = "%s keeps its data in region %s. this version of carlos cannot read it from there. run carlos update, then try again."
)

// forwardable is whether this console forwards reads of rec: it can sign,
// and rec is in a named platform region (a fleet placement keeps its own
// sentence, and a home-region record seen from a regional console has no
// region to name).
func (s *Server) forwardable(rec provision.Record) bool {
	_, fleet := registry.FleetPlacement(rec.Region)
	_, ok := s.dbForwardSigner()
	return ok && !fleet && dbreads.ValidRegion(rec.Region)
}

// dbResidency answers a record homed outside this console's region (spec
// §5.5 step 1, §9.2). Not forwardable: main's refusal, byte for byte.
// Forwardable: DBF-1 or DBF-2 on complete evidence, else the forward marker,
// whose sentence an old CLI prints (DBF-14).
func (s *Server) dbResidency(w http.ResponseWriter, r *http.Request, rec provision.Record) {
	if !s.forwardable(rec) {
		jsonError(w, http.StatusConflict, s.dbResidencyRefusal(rec))
		return
	}
	snap, err := s.liveEdges(r.Context())
	switch dbForwardEvidence(snap, err, rec.Region) {
	case dbForwardNone:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoCapable, rec.Host, rec.Region))
		return
	case dbForwardDead:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoneLive, rec.Host, rec.Region))
		return
	}
	writeJSON(w, http.StatusConflict, dbdoor.ForwardBody{
		Error: fmt.Sprintf(dbfUpdateCarlos, rec.Host, rec.Region), Forward: 1,
		Region: rec.Region, ForwardOps: []string{dbreads.OpQuery},
	})
}
```

In `internal/console/apidb.go`, `dbGate`'s `ErrResidency` case becomes:

```go
	case errors.Is(err, ErrResidency):
		s.dbResidency(w, r, src.rec)
		return dbRequest{}, false
```

Add `"DBF-1", "DBF-2", "DBF-14"` to `dbForwardCopyBuilt`.

- [ ] **Step 4: Run the tests and the gate**

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

- [ ] **Step 5: Commit**

```bash
git add internal/console approved_db_forward_copy_test.go
git commit -m "console: the db door marks a capable cross-region instance (DBF-1, DBF-2, DBF-14)"
```

---

### Task 6: The reads route: the gate, admission, the signed request

Waits for Task 1 (DBF-9, DBF-10, DBF-21, DBF-22 approved).

**Files:**
- Modify: `internal/console/apidbforward.go` (the handler and its admission half)
- Modify: `internal/console/authcli.go:751` (register the route after the two db door routes)
- Modify: `internal/console/testdata/claim-routes.txt` (one line, in sorted position)
- Modify: `internal/console/apidbforward_test.go`
- Modify: `approved_db_forward_copy_test.go` (`dbForwardCopyBuilt` gains DBF-9, DBF-10, DBF-21, DBF-22)

**Interfaces:**
- Consumes: `dbForwardSigner`, `forwardable`, `dbForwardEvidence`, `s.dbReads`, `s.dbf` (Tasks 3 to 5); `resolveDBSource`, `dbClassRefusal`, `dbResidencyRefusal`, `apiAccountApp`, `principalFrom`, `s.cli.db.allow`, `hashToken`, `bearer`, `readJSONLimit`, `jsonError`, `s.store.RecordActivityContext`, `release.BumpSerial`, `release.SerialKey`, `dbAuditAction` (existing console code); `dbreads.Sign`, `MintID`, `RequestKey`, `QueueKey`, `ParseReplyKey`, `ReadBody`; `dbquery.CheckSQL`; `hostclaim.Incarnation`.
- Produces: `(*Server).handleAPIDBRead(w, r)`; `(*Server).admitDBRead(w, r) (dbReadAdmitted, string, bool)`; `type dbReadAdmitted struct{ id, region string; issued time.Time }`; `dbForwardAuditDetail(region, sqlText string, at time.Time) string`; `(*Server).cancelDBRead(id)`; `(*Server).announceDBRead(ctx)`; `dbfTooMany`, `dbfMoving`, `dbfAudit`, `dbfAuditAt`; the stream half (`streamDBRead`) is Task 7's; test helpers `(*dbfFixture).readURL(host)`, `queryBody(t, sql)`, `(*dbfFixture).openRead(t, host, bearer, body) *http.Response`, `readLines(t, body io.Reader) []dbreads.Line`, `refusal(t, res)`, `dbfPutHook`, `dbReadAudited(t, st)`, `(*dbfFixture).dbfServe(t, ctx, sql)`, `dbfBucket(t, s)`, `dbfCtxStore`; the seam `s.dbf.beforeWrite`.

- [ ] **Step 1: Write the failing tests**

Append to `internal/console/apidbforward_test.go`:

```go
func (fx *dbfFixture) readURL(host string) string {
	return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + host + "/reads"
}

func queryBody(t *testing.T, sqlText string) (dbreads.ReadBody, *ecdh.PrivateKey) {
	t.Helper()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		t.Fatal(err)
	}
	return dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}, key
}

// openRead posts a read and returns the open response; the caller reads its
// lines or closes it.
func (fx *dbfFixture) openRead(t *testing.T, host, bearer string, body dbreads.ReadBody) *http.Response {
	t.Helper()
	raw, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, fx.readURL(host), bytes.NewReader(raw))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+bearer)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	return res
}

func readLines(t *testing.T, r io.Reader) []dbreads.Line {
	t.Helper()
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
	var out []dbreads.Line
	for sc.Scan() {
		var l dbreads.Line
		if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
			t.Fatalf("not a line: %s", sc.Bytes())
		}
		out = append(out, l)
	}
	return out
}

func refusal(t *testing.T, res *http.Response) (int, string) {
	t.Helper()
	defer res.Body.Close()
	b, _ := io.ReadAll(res.Body)
	return res.StatusCode, string(b)
}

func TestDBReadRouteRefusesBeforeAdmitting(t *testing.T) {
	fx := newDBFFixture(t, true)
	good, _ := queryBody(t, "SELECT 1")
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "here.example.com", Region: "ie", CreatedAt: time.Now().UTC()})
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "moving.example.com", Region: "uk", CreatedAt: time.Now().UTC(),
		Move: &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}})
	two := good
	two.SQL = "SELECT 1; SELECT 2"
	pull := good
	pull.Op, pull.SQL = dbreads.OpPull, ""
	badKey := good
	badKey.ReplyKey = "AAAA"
	future := good
	future.At = time.Now().Add(time.Hour).UTC().Format(time.RFC3339)
	for name, c := range map[string]struct {
		host   string
		body   dbreads.ReadBody
		status int
		want   string
	}{
		"an in-region instance": {"here.example.com", good, http.StatusNotFound, ""},
		"no such instance":      {"nope.example.com", good, http.StatusNotFound, ""},
		"moving":                {"moving.example.com", good, http.StatusConflict, fmt.Sprintf(dbfMoving, "moving.example.com")},
		"two statements":        {dbfHost, two, http.StatusBadRequest, dbquery.ErrOneStatement.Error()},
		"an op this console does not forward": {dbfHost, pull, http.StatusBadRequest, ""},
		"a reply key that is not a key":       {dbfHost, badKey, http.StatusBadRequest, ""},
		"a time in the future":                {dbfHost, future, http.StatusBadRequest, ""},
	} {
		status, body := refusal(t, fx.openRead(t, c.host, fx.member, c.body))
		if status != c.status || !strings.Contains(body, c.want) {
			t.Errorf("%s: %d %s", name, status, body)
		}
	}
	// SQL that JSON decoding would have rewritten (plan review finding 36):
	// a byte that is not UTF-8, and a lone surrogate escape.
	for name, sqlJSON := range map[string]string{
		"a byte that is not UTF-8": "\"SELECT '\xff'\"",
		"a lone surrogate":         `"SELECT '\ud800'"`,
	} {
		raw := `{"op":"query","sql":` + sqlJSON + `,"reply_key":"` + good.ReplyKey + `"}`
		req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(raw))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Authorization", "Bearer "+fx.member)
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		if status, body := refusal(t, res); status != http.StatusBadRequest {
			t.Errorf("%s: %d %s", name, status, body)
		}
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a refused read wrote %v", keys)
	}
	fx.srv.cfg.DBForward = false
	status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in region ie.") {
		t.Fatalf("forwarding off: %d %s", status, body)
	}
	fx.srv.cfg.DBForward, fx.srv.cfg.HostClaim = true, true
	status, body = refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in") {
		t.Fatalf("a host-claim console: %d %s", status, body)
	}
}

func TestDBReadRouteRefusesOnCompleteEvidence(t *testing.T) {
	fx := newDBFFixture(t, true)
	putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false)
	good, _ := queryBody(t, "SELECT 1")
	status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) {
		t.Fatalf("%d %s", status, body)
	}
}

// Admission, then the request, the queue entry, the serial and the audit
// row, all before the 200.
func TestDBReadQueuesASignedRequestAndAudits(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.pickup = 200 * time.Millisecond
	body, _ := queryBody(t, "SELECT count(*) FROM orders")
	res := fx.openRead(t, dbfHost, fx.member, body)
	lines := readLines(t, res.Body)
	res.Body.Close()
	if res.StatusCode != http.StatusOK || len(lines) == 0 || lines[0].Admitted == "" {
		t.Fatalf("%d %+v", res.StatusCode, lines)
	}
	id := lines[0].Admitted
	raw, _, err := fx.rel.Get(context.Background(), dbreads.RequestKey(id))
	if err != nil {
		t.Fatal(err)
	}
	req, err := dbreads.Verify(fx.pub, raw, id, "uk", time.Now())
	if err != nil || req.SQL != body.SQL || req.Host != dbfHost || req.By != "member@example.com" || req.Account != release.OpsAccount {
		t.Fatalf("request %+v: %v", req, err)
	}
	issued, _ := req.Issued()
	if _, _, err := fx.rel.Get(context.Background(), dbreads.QueueKey("uk", issued, id)); err != nil {
		t.Fatalf("no queue entry: %v", err)
	}
	if _, _, err := fx.rel.Get(context.Background(), release.SerialKey); err != nil {
		t.Fatalf("the serial was not bumped: %v", err)
	}
	evs, _ := fx.st.Events(OpsAccountID, "shop", 10)
	want := dbForwardAuditDetail("uk", body.SQL, time.Time{})
	found := false
	for _, e := range evs {
		found = found || (e.Action == "db-read" && e.Target == dbfHost && e.Detail == want)
	}
	if !found || strings.Contains(want, "orders") {
		t.Fatalf("audit %q in %+v", want, evs)
	}
}

// dbfPutHook runs hook after each create of a key with suffix, once the
// create has landed.
type dbfPutHook struct {
	release.Store
	suffix string
	hook   func()
}

func (h *dbfPutHook) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := h.Store.Put(ctx, key, body, ifMatch)
	if strings.HasSuffix(key, h.suffix) || (h.suffix == "queue" && strings.HasPrefix(key, dbreads.Prefix+"queue/")) {
		h.hook()
	}
	return etag, err
}

// dbReadAudited is safe from a handler's goroutine: it never calls t.Fatal.
func dbReadAudited(t *testing.T, st *Store) bool {
	t.Helper()
	evs, err := st.Events(OpsAccountID, "shop", 10)
	if err != nil {
		t.Error(err)
		return false
	}
	for _, e := range evs {
		if e.Action == "db-read" && e.Target == dbfHost {
			return true
		}
	}
	return false
}

// dbfServe drives the handler in process with ctx as the request's own
// context, so a test can end the request at an exact step.
func (fx *dbfFixture) dbfServe(t *testing.T, ctx context.Context, sqlText string) *httptest.ResponseRecorder {
	t.Helper()
	body, _ := queryBody(t, sqlText)
	raw, _ := json.Marshal(body)
	req := httptest.NewRequest(http.MethodPost, fx.readURL(dbfHost), bytes.NewReader(raw)).WithContext(ctx)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+fx.member)
	rec := httptest.NewRecorder()
	fx.srv.ServeHTTP(rec, req)
	return rec
}

// dbfBucket is what a read left under db-reads/: request objects, queue
// entries and cancels, by count.
func dbfBucket(t *testing.T, s release.Store) (requests, queued, cancels int) {
	t.Helper()
	keys, _ := s.List(context.Background(), dbreads.Prefix)
	for _, k := range keys {
		switch {
		case strings.HasPrefix(k, dbreads.Prefix+"queue/"):
			queued++
		case strings.HasSuffix(k, "/request.json"):
			requests++
		case strings.HasSuffix(k, "/cancel"):
			cancels++
		}
	}
	return
}

// dbfCtxStore refuses a write whose context has ended, as the S3 store
// does and DirStore does not, and runs onGet once on a read of key.
type dbfCtxStore struct {
	release.Store
	key   string
	onGet func()
	once  sync.Once
}

func (c *dbfCtxStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == c.key {
		c.once.Do(c.onGet)
	}
	return c.Store.Get(ctx, key)
}

func (c *dbfCtxStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if err := ctx.Err(); err != nil {
		return "", err
	}
	return c.Store.Put(ctx, key, body, ifMatch)
}

// Plan review findings 11, 28 and 33 (J1, spec §5.5 step 3 and §9.3 rev 6):
// the request's context is checked immediately before every write of
// admission. Ended before the audit row: nothing at all. Before
// request.json: the row only. Before the queue entry: the row, request.json
// and the cancel. Before the serial: the row, both objects and the cancel.
// Never a serial bump, never a 200. And the bump itself runs on the
// request's context, so a request that ends while it reads the serial does
// not write it.
// Mutations, each red in its own case: drop any one of the four ended()
// checks; drop the cancel on a published step; write the audit row after
// the creates ("the audit row is down before the request"); call
// s.bumpSerial instead of announceDBRead ("ended during the serial read":
// the detached write lands).
func TestDBReadEndedDuringAdmissionWritesOnlyTheCancel(t *testing.T) {
	t.Run("the audit row is down before the request", func(t *testing.T) {
		fx := newDBFFixture(t, true)
		auditedFirst := false
		fx.srv.cfg.Releases = &dbfPutHook{Store: fx.rel, suffix: "/request.json", hook: func() {
			auditedFirst = dbReadAudited(t, fx.st)
		}}
		fx.srv.dbf.pickup = 100 * time.Millisecond
		body, _ := queryBody(t, "SELECT 1")
		res := fx.openRead(t, dbfHost, fx.member, body)
		io.Copy(io.Discard, res.Body)
		res.Body.Close()
		if !auditedFirst {
			t.Fatal("request.json was written before the audit row")
		}
	})
	for _, c := range []struct {
		step                       string
		audited                    bool
		requests, queued, cancels int
	}{
		{"audit", false, 0, 0, 0},
		{"request", true, 0, 0, 0},
		{"queue", true, 1, 0, 1},
		{"serial", true, 1, 1, 1},
	} {
		t.Run("ended before "+c.step, func(t *testing.T) {
			fx := newDBFFixture(t, true)
			_, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey)
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			fx.srv.dbf.beforeWrite = func(step string) {
				if step == c.step {
					cancel()
				}
			}
			rec := fx.dbfServe(t, ctx, "SELECT 1")
			// The ended path writes no response (nobody is there to read
			// it), and a recorder never written reports 200: read the body.
			if strings.Contains(rec.Body.String(), "admitted") {
				t.Fatalf("an ended request was admitted: %s", rec.Body)
			}
			if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore {
				t.Fatal("the serial was bumped after the request ended")
			}
			requests, queued, cancels := dbfBucket(t, fx.rel)
			if got := dbReadAudited(t, fx.st); got != c.audited || requests != c.requests || queued != c.queued || cancels != c.cancels {
				t.Fatalf("audited %v, requests %d, queue entries %d, cancels %d", got, requests, queued, cancels)
			}
		})
	}
	t.Run("ended during the serial read", func(t *testing.T) {
		fx := newDBFFixture(t, true)
		_, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey)
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		fx.srv.cfg.Releases = &dbfCtxStore{Store: fx.rel, key: release.SerialKey, onGet: cancel}
		fx.dbfServe(t, ctx, "SELECT 1")
		if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore {
			t.Fatal("the serial was written after the request ended during the bump")
		}
	})
}

// Plan review finding 13: a body delivered slowly is cut at s.dbf.preHeader,
// well inside backhaul's 30 s header wait, and admits nothing.
// Mutation: drop the preHeader read deadline, keeping only the bound's (red:
// the connection is still open at 3 s).
func TestDBReadStalledBodyIsCutAtThePreHeaderBound(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.preHeader = 300 * time.Millisecond
	conn, err := net.Dial("tcp", strings.TrimPrefix(fx.ts.URL, "http://"))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	path := strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL)
	fmt.Fprintf(conn, "POST %s HTTP/1.1\r\nHost: console\r\nAuthorization: Bearer %s\r\n"+
		"Content-Type: application/json\r\nContent-Length: 4096\r\n\r\n{\"op\":", path, fx.member)
	conn.SetReadDeadline(time.Now().Add(3 * time.Second))
	start := time.Now()
	_, err = io.ReadAll(conn) // a refusal, then the close; or the close alone
	if ne, ok := err.(net.Error); ok && ne.Timeout() {
		t.Fatalf("the stalled body still held the request after %s", time.Since(start))
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a stalled body wrote %v", keys)
	}
}
```

Add `"bufio"`, `"bytes"`, `"crypto/ecdh"`, `"io"`, `"net"`, `"net/http"`, `"net/http/httptest"`, `"sync"` and the `dbquery` import.

The admission bounds test (DBF-9, T15) is Task 7's: it needs the stream to hold its requests open (plan review finding 7).

- [ ] **Step 2: Run them to verify they fail**

Run: `go test ./internal/console/ -run 'TestDBReadRoute|TestDBReadQueues|TestDBReadEndedDuringAdmission|TestDBReadStalledBody' -count=1`
Expected: FAIL: the route answers 404 for everything.

- [ ] **Step 3: Write the admission half**

Append to `internal/console/apidbforward.go` (imports `context`, `crypto/sha256`, `encoding/hex`, `errors`, `log`, `os`, `dbquery`, `hostclaim`):

```go
// The forwarded read's other console sentences (copy record DBF-9, DBF-10,
// DBF-21, DBF-22).
const (
	dbfTooMany = "too many database reads are open. wait a minute and try again."
	dbfMoving  = "%s is moving between regions. read its database once the move has finished."
	dbfAudit   = "forwarded a database query to region %s. the query was %d bytes, sha256 %s (cli)"
	dbfAuditAt = "forwarded a database query to region %s, as of %s. the query was %d bytes, sha256 %s (cli)"
)

const dbReadRetryAfter = "30"

// dbForwardAuditDetail is the db-read row's detail (spec §9.3): the SQL's
// length and the first 16 hex digits of its SHA-256, never its text.
func dbForwardAuditDetail(region, sqlText string, at time.Time) string {
	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)
}

// dbReadAdmitted is a read this console has admitted, signed and queued.
type dbReadAdmitted struct {
	id, region string
	issued     time.Time
}

// handleAPIDBRead is POST /api/cli/apps/{account}/{app}/db/{host}/reads
// (spec §9.1): one forwarded read, which exists only inside this request
// (§5.2, J1). The admission counts are released by this handler's return and
// nowhere else (J5).
func (s *Server) handleAPIDBRead(w http.ResponseWriter, r *http.Request) {
	rc := http.NewResponseController(w)
	// Everything before the 200, the body included, gets s.dbf.preHeader:
	// a body delivered slowly must not hold the request past backhaul's
	// 30 s header wait (plan review finding 13).
	if err := rc.SetReadDeadline(time.Now().Add(s.dbf.preHeader)); err != nil && !errors.Is(err, http.ErrNotSupported) {
		log.Printf("console: db read: set the admission read deadline: %v", err)
	}
	a, actor, ok := s.admitDBRead(w, r)
	if !ok {
		return
	}
	defer s.dbReads.release(actor)
	// Then the server's ReadTimeout would end this connection under a read
	// that lasts longer; the read's own bound holds it instead, as the
	// import door's read deadline does (apiinstanceimport.go).
	if err := rc.SetReadDeadline(time.Now().Add(s.dbf.bound + time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) {
		log.Printf("console: db read: extend the read deadline: %v", err)
	}
	s.streamDBRead(w, r, rc, a)
}

// cancelDBRead is the one write J1 allows after a read's request ends
// without a result: best effort, so the owner stops early and a box that
// has not picked it skips it.
func (s *Server) cancelDBRead(id string) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := s.cfg.Releases.Put(ctx, dbreads.CancelKey(id), nil, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
		log.Printf("console: db read %s: cancel: %v", id, err)
	}
}

// announceDBRead bumps the serial on the request's own context, so a
// request that has already ended announces nothing (J1; plan review finding
// 11). bumpSerial detaches on purpose, for callers whose writes outlive
// them; this read does not outlive its request.
func (s *Server) announceDBRead(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	if err := release.BumpSerial(ctx, s.cfg.Releases, time.Now()); err != nil && ctx.Err() == nil {
		log.Printf("console: db read: serial bump: %v (boxes list every 30 s anyway)", err)
	}
}

// admitDBRead is spec §5.5 step 3, all of it bounded by s.dbf.preHeader so
// the 200 leaves well inside backhaul's 30 s header wait: dbGate's order
// (person, membership with the byte-identical 404, the db budget,
// configuration, the record, the class rail, pooled, fleet), then a moving
// record, the SQL guard, capability on complete evidence, the admission
// counts, and the signed request, its queue entry, the serial and the audit
// row. Any refusal is an ordinary status and JSON error.
func (s *Server) admitDBRead(w http.ResponseWriter, r *http.Request) (dbReadAdmitted, string, bool) {
	none := dbReadAdmitted{}
	ctx, cancel := context.WithTimeout(r.Context(), s.dbf.preHeader)
	defer cancel()
	if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
		http.NotFound(w, r)
		return none, "", false
	}
	actor, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return none, "", false
	}
	if !s.cli.db.allow(hashToken(bearer(r)), time.Now()) {
		jsonError(w, http.StatusTooManyRequests, "too many database requests. Wait a minute and try again.")
		return none, "", false
	}
	if s.cfg.Releases == nil || s.cfg.HibernateBucket == "" {
		jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot read databases.")
		return none, "", false
	}
	var body dbreads.ReadBody
	if !readJSONLimit(w, r, &body, dbreads.MaxSQLBytes+8<<10) {
		return none, "", false
	}
	host := r.PathValue("host")
	src, err := s.resolveDBSource(ctx, acct, actor, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost):
		http.NotFound(w, r)
		return none, "", false
	case errors.Is(err, errSourceClass):
		jsonError(w, http.StatusConflict, dbClassRefusal(src.rec))
		return none, "", false
	case errors.Is(err, ErrPooled):
		jsonError(w, http.StatusConflict, fmt.Sprintf(
			"%s runs in the pool %s, so its data is on the pool's own server. carlos db cannot read it.", host, src.rec.Pool))
		return none, "", false
	case errors.Is(err, ErrResidency):
		// The case this route exists for.
	case err == nil:
		// In region: the db door serves it directly. There is no forwarded
		// read for it until in-region reads move to the box path (plan 4).
		http.NotFound(w, r)
		return none, "", false
	case errors.Is(err, errSourceResolve):
		log.Printf("console: db read %s/%s/%s: resolve: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
		return none, "", false
	default:
		log.Printf("console: db read %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
		return none, "", false
	}
	rec := src.rec
	signer, ok := s.dbForwardSigner()
	if !ok || !s.forwardable(rec) {
		jsonError(w, http.StatusConflict, s.dbResidencyRefusal(rec))
		return none, "", false
	}
	if rec.Move != nil {
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfMoving, host))
		return none, "", false
	}
	if body.Op != dbreads.OpQuery {
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	// encoding/json turns bytes that are not UTF-8, and a lone surrogate
	// escape, into U+FFFD without a word, so the SQL decoded here may not be
	// the SQL sent. The box must run exactly the bytes typed or be told no
	// (plan 1's Review Focus 1), so any U+FFFD is refused; a query that
	// meant one literally is refused too, and says so the same way (plan
	// review finding 36). The CLI refuses first, with DBF-32.
	if strings.ContainsRune(body.SQL, utf8.RuneError) {
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	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
	}
	var at time.Time
	if body.At != "" {
		if at, err = time.Parse(time.RFC3339, body.At); err != nil || at.After(time.Now().Add(dbreads.Skew)) {
			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
			return none, "", false
		}
	}
	if _, err := dbreads.ParseReplyKey(body.ReplyKey); err != nil {
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	snap, serr := s.liveEdges(ctx)
	switch dbForwardEvidence(snap, serr, rec.Region) {
	case dbForwardNone:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoCapable, host, rec.Region))
		return none, "", false
	case dbForwardDead:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoneLive, host, rec.Region))
		return none, "", false
	}
	if !s.dbReads.acquire(actor) {
		w.Header().Set("Retry-After", dbReadRetryAfter)
		jsonError(w, http.StatusTooManyRequests, dbfTooMany)
		return none, "", false
	}
	admitted := false
	defer func() {
		if !admitted {
			s.dbReads.release(actor)
		}
	}()
	fail := func(what string, err error) (dbReadAdmitted, string, bool) {
		log.Printf("console: db read %s/%s/%s: %s: %v", acct.Sqid, app, host, what, err)
		jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
		return none, "", false
	}
	id, err := dbreads.MintID()
	if err != nil {
		return fail("mint id", err)
	}
	inc, err := hostclaim.Incarnation(acct.Sqid, app, host, rec.CreatedAt)
	if err != nil {
		return fail("incarnation", err)
	}
	issued := time.Now().UTC().Truncate(time.Second)
	atRaw := ""
	if !at.IsZero() {
		atRaw = at.UTC().Format(time.RFC3339Nano)
	}
	wire, err := dbreads.Sign(signer, dbreads.Request{ID: id, Account: acct.Sqid, App: app, Host: host, Incarnation: inc,
		Region: rec.Region, Op: body.Op, SQL: body.SQL, At: atRaw, ReplyKey: body.ReplyKey, By: actor,
		IssuedAt: issued.Format(time.RFC3339)})
	if err != nil {
		// SQL the envelope cannot carry intact (invalid UTF-8).
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	// J1 at every write (plan review finding 28): the request's context is
	// checked immediately before each one, because the stores do not stop a
	// write for it (DirStore.Put ignores its context,
	// internal/release/store.go:414). The audit row comes first, while the
	// request is open and before any box can see the read, so a read a box
	// runs always has its row (spec §9.3 rev 6). A request that ends before
	// request.json is written leaves at most that row; one that ends after
	// leaves request.json, perhaps the queue entry, and the cancel, never a
	// serial bump. A row the store cannot write refuses the read rather than
	// run it unaudited.
	ended := func(step string, published bool) bool {
		if s.dbf.beforeWrite != nil {
			s.dbf.beforeWrite(step)
		}
		if ctx.Err() == nil {
			return false
		}
		if published {
			s.cancelDBRead(id)
		}
		return true
	}
	if ended("audit", false) {
		return none, "", false
	}
	if err := s.store.RecordActivityContext(ctx, acct.ID, app, actor, Activity{Action: dbAuditAction,
		TargetType: "instance", Target: host, Detail: dbForwardAuditDetail(rec.Region, body.SQL, at)}, time.Now()); err != nil {
		return fail("audit", err)
	}
	if ended("request", false) {
		return none, "", false
	}
	if _, err := s.cfg.Releases.Put(ctx, dbreads.RequestKey(id), wire, ""); err != nil {
		// An ambiguous create may have landed: cancel, as for the queue.
		s.cancelDBRead(id)
		return fail("write the request", err)
	}
	if ended("queue", true) {
		return none, "", false
	}
	if _, err := s.cfg.Releases.Put(ctx, dbreads.QueueKey(rec.Region, issued, id), nil, ""); err != nil {
		// The queue entry may have landed on an ambiguous answer: cancel, so
		// a box that lists it skips it (J1's one write).
		s.cancelDBRead(id)
		return fail("write the queue entry", err)
	}
	if ended("serial", true) {
		return none, "", false
	}
	// The bump runs on the request's context, not bumpSerial's detached one,
	// so a request that ends while it runs does not announce (finding 33).
	s.announceDBRead(ctx)
	admitted = true
	return dbReadAdmitted{id: id, region: rec.Region, issued: issued}, actor, true
}
```

Add `"strings"` and `"unicode/utf8"` to `apidbforward.go`'s imports if they are not there yet.

Until Task 7, add a stub so this task builds, and delete it in Task 7:

```go
// streamDBRead is Task 7's.
func (s *Server) streamDBRead(w http.ResponseWriter, _ *http.Request, _ *http.ResponseController, a dbReadAdmitted) {
	w.Header().Set("Content-Type", "application/x-ndjson")
	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(dbreads.Line{Admitted: a.id})
}
```

Register the route in `registerCLIRoutes`, after the two db door routes:

```go
	// The forwarded read (db region forward spec §9.1): people only, the db
	// door's whole gate once, then the read inside the same request.
	s.cliRoute("POST /api/cli/apps/{account}/{app}/db/{host}/reads", svcrole.Set{}, s.handleAPIDBRead)
```

and add `route POST /api/cli/apps/{account}/{app}/db/{host}/reads` to `internal/console/testdata/claim-routes.txt` in its sorted position (after the `route POST /api/cli/apps/{account}/{app}/` lines that sort before it; run the claim-routes test to confirm the place).

Add `"DBF-9", "DBF-10", "DBF-21", "DBF-22"` to `dbForwardCopyBuilt`.

- [ ] **Step 4: Run the tests**

Run: `go test ./internal/console/ -run 'TestDBReadRoute|TestDBReadQueues|TestDBReadEndedDuringAdmission|TestDBReadStalledBody|TestClaim.*Routes' -count=1 && go test . -run 'TestDBForward' -count=1`
Expected: PASS. (`TestClaim.*Routes` is the claim-routes census; list the exact name with `grep -n 'claim-routes.txt' internal/console/*_test.go` if the pattern matches nothing.)

- [ ] **Step 5: Commit**

```bash
git add internal/console approved_db_forward_copy_test.go
git commit -m "console: the reads route's gate and admission (DBF-9, DBF-10, DBF-21, DBF-22)"
```

---

### Task 7: The reads route: the stream

**Files:**
- Modify: `internal/console/apidbforward.go` (replace the stub with the stream)
- Modify: `internal/console/apidbforward_test.go`

**Interfaces:**
- Consumes: `dbReadAdmitted`, `s.dbf` (Tasks 3, 6); `dbreads.Line`, `Picked`, `Result`, `PickedKey`, `ResultKey`, `OutPrefix`, `ParseChunkKey`, `ChunkKey`, `CancelKey`, the wait and ended words; `dbreadstest.Start`, `Box.Agent` (plan 1).
- Consumes also: `(*Server).cancelDBRead` (Task 6), `s.dbf.call` (Task 3).
- Produces: `(*Server).streamDBRead(w, r, rc, a)`; `type dbLineWriter` (with `until`), `(*dbLineWriter).tick(ctx, phase, poll, every) error`; `errDBClientGone`; `dbfEndSlack`; `type dbReadRelay`; `dbfCall[T](st, ctx, phase, op) (T, error)`; `(*Server).readDBResult(ctx, id) *dbreads.Result`; test helpers `runBox(t, box)`, `(*dbfFixture).startBox(t, dir) *dbreadstest.Box`, `dbfOpen`, `(*dbfFixture).openHeld`, `(*dbfFixture).admitWithin`, `refusedTooMany`, `slowStore`, `scriptedBox`, `dbReadHookStore`, `dbfHideStore`, `dbfHeldWriter`, `dbfDeadlineListener`, `dbfDeadlineConn`, `strictStore`.

- [ ] **Step 1: Write the failing tests**

Append to `internal/console/apidbforward_test.go`:

```go
// startBox runs plan 1's real agent and worker code in process for uk,
// passing every 20 ms: a stand-in for the box's serial poll, not acceptance.
func (fx *dbfFixture) startBox(t *testing.T, replicaDir string) *dbreadstest.Box {
	t.Helper()
	box := dbreadstest.Start(t, dbreadstest.Options{Store: fx.rel, Region: "uk", Box: "uk-1", Pub: fx.pub, HibernatePrefix: "instances",
		OpenReplica: func(context.Context, string) (dbdoor.ReplicaReader, error) { return lsfile.NewReplicaClient(replicaDir), nil }})
	runBox(t, box)
	return box
}

func runBox(t *testing.T, box *dbreadstest.Box) {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		for ctx.Err() == nil {
			box.Agent.Pass(ctx)
			time.Sleep(20 * time.Millisecond)
		}
	}()
	t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() })
}

// openAnswer opens a done result's chunks and summary with the CLI's key.
func openAnswer(t *testing.T, key *ecdh.PrivateKey, lines []dbreads.Line) ([]byte, dbreads.Summary) {
	t.Helper()
	id := lines[0].Admitted
	res := lines[len(lines)-1].Result
	if res == nil || res.State != dbreads.StateDone {
		t.Fatalf("no done result: %+v", lines[len(lines)-1])
	}
	var rows bytes.Buffer
	for _, l := range lines {
		if l.Chunk == 0 {
			continue
		}
		plain, err := dbreads.OpenChunk(key, id, l.Chunk, dbreads.StreamRows, l.Chunk == res.LastSeq, l.Data)
		if err != nil {
			t.Fatalf("chunk %d: %v", l.Chunk, err)
		}
		rows.Write(plain)
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		t.Fatal(err)
	}
	return rows.Bytes(), s
}

func TestDBReadStreamsAQueryEndToEnd(t *testing.T) {
	fx := newDBFFixture(t, true)
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	fx.startBox(t, dir)
	body, key := queryBody(t, "SELECT count(*) AS n FROM t")
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK || res.Header.Get("Content-Type") != "application/x-ndjson" ||
		res.Header.Get("Cache-Control") != "no-store" || res.ContentLength != -1 {
		t.Fatalf("%d %v", res.StatusCode, res.Header)
	}
	lines := readLines(t, res.Body)
	if lines[0].Admitted == "" || lines[1].Picked != "uk-1" {
		t.Fatalf("lines %+v", lines)
	}
	rows, _ := openAnswer(t, key, lines)
	if string(rows) != "n\n30\n" {
		t.Fatalf("rows %q", rows)
	}
}

// No box: wait lines while the console waits, then ended not-picked after
// the pickup window, a cancel, and the slot free again.
func TestDBReadEndsNotPickedAndCancels(t *testing.T) {
	fx := newDBFFixture(t, true)
	// issued_at is truncated to the second (production's semantics), so the
	// pickup window seen from admission is pickup minus up to a second: 1.4 s
	// leaves at least 400 ms, eight wait periods (plan review finding 30).
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, 1400*time.Millisecond
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	lines := readLines(t, res.Body)
	res.Body.Close()
	waits := 0
	for _, l := range lines {
		if l.Wait == dbreads.WaitPickup {
			waits++
		}
	}
	last := lines[len(lines)-1]
	if waits < 3 || last.Ended != dbreads.EndedNotPicked || last.Result != nil {
		t.Fatalf("lines %+v", lines)
	}
	if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(lines[0].Admitted)); err != nil {
		t.Fatalf("no cancel: %v", err)
	}
}

// T7, the console's half: admitted leaves with the headers, and every line
// is flushed as it is written.
func TestDBReadLinesAreFlushedAsWritten(t *testing.T) {
	fx := newDBFFixture(t, true)
	// At least a second of pickup is left after issued_at's truncation: room
	// for the five lines below at 100 ms each (plan review finding 30).
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 100*time.Millisecond, 2*time.Second
	body, _ := queryBody(t, "SELECT 1")
	start := time.Now()
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	prev := start
	for i := 0; i < 5; i++ {
		if _, err := br.ReadBytes('\n'); err != nil {
			t.Fatal(err)
		}
		if gap := time.Since(prev); gap > 400*time.Millisecond {
			t.Fatalf("line %d arrived %s after the last: not flushed", i, gap)
		}
		prev = time.Now()
	}
}

// dbfOpen is one open read: its response, a reader over its lines, and its
// id ("" when it was refused).
type dbfOpen struct {
	res *http.Response
	br  *bufio.Reader
	id  string
}

func (fx *dbfFixture) openHeld(t *testing.T, base, bearer string) dbfOpen {
	t.Helper()
	body, _ := queryBody(t, "SELECT 1")
	raw, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, base+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfHost+"/reads", bytes.NewReader(raw))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+bearer)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	o := dbfOpen{res: res, br: bufio.NewReader(res.Body)}
	if res.StatusCode == http.StatusOK {
		line, _ := o.br.ReadBytes('\n')
		var l dbreads.Line
		json.Unmarshal(line, &l)
		o.id = l.Admitted
	}
	return o
}

// untilPicked reads lines until the console has written picked, so the
// handler is in the owner's phase.
func (o dbfOpen) untilPicked(t *testing.T) {
	t.Helper()
	for {
		line, err := o.br.ReadBytes('\n')
		if err != nil {
			t.Fatalf("%s ended before picked: %v", o.id, err)
		}
		var l dbreads.Line
		json.Unmarshal(line, &l)
		if l.Picked != "" {
			return
		}
	}
}

func refusedTooMany(t *testing.T, o dbfOpen, what string) {
	t.Helper()
	status, body := refusal(t, o.res)
	if status != http.StatusTooManyRequests || o.res.Header.Get("Retry-After") == "" || !strings.Contains(body, dbfTooMany) {
		t.Fatalf("%s: %d %s", what, status, body)
	}
}

// admitWithin retries until a read is admitted: a handler frees its slot when
// it notices its client went, which takes up to a poll.
func (fx *dbfFixture) admitWithin(t *testing.T, base, bearer string) dbfOpen {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		o := fx.openHeld(t, base, bearer)
		if o.id != "" {
			return o
		}
		o.res.Body.Close()
		if time.Now().After(deadline) {
			t.Fatal("a disconnected read still holds its slot")
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// DBF-9 and T15 (J5; plan review findings 7 and 18): two open reads per
// person and sixteen per console, held by the open requests. Both of a
// person's slots come back when both CLIs leave before pickup, and again
// when both leave after it; the seventeenth read across nine people is
// refused; each disconnected read writes cancel; and a console restarted
// under open reads recovers nothing and starts with empty counts.
// Mutations: drop handleAPIDBRead's defer release (red: the second pair is
// never admitted); release at the end of admitDBRead (red: the third read of
// a pair is admitted); skip cancelDBRead on a client that went (red: no
// cancel); a startup pass that writes under db-reads/ (red: the listing
// changes across NewServer).
func TestDBReadAdmissionBounds(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, time.Minute
	base := fx.ts.URL
	var left []string

	// Before pickup.
	a, b := fx.openHeld(t, base, fx.member), fx.openHeld(t, base, fx.member)
	if a.id == "" || b.id == "" {
		t.Fatal("the first two reads were refused")
	}
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read")
	a.res.Body.Close()
	b.res.Body.Close()
	left = append(left, a.id, b.id)

	// Both slots come back; the pair is picked, then both CLIs leave.
	c, d := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member)
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the first pair left")
	for _, o := range []dbfOpen{c, d} {
		scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Minute))
		o.untilPicked(t)
		o.res.Body.Close()
		left = append(left, o.id)
	}
	e, f := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member)
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the picked pair left")
	e.res.Body.Close()
	f.res.Body.Close()
	left = append(left, e.id, f.id)
	for _, id := range left {
		deadline := time.Now().Add(5 * time.Second)
		for {
			if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(id)); err == nil {
				break
			}
			if time.Now().After(deadline) {
				t.Fatalf("%s: no cancel after its CLI left", id)
			}
			time.Sleep(20 * time.Millisecond)
		}
	}

	// Sixteen per console, across people.
	var held []dbfOpen
	people := dbreads.OpenPerConsole/dbreads.OpenPerPerson + 1
	for i := 0; i < people; i++ {
		tok := fx.member
		if i > 0 {
			addr := fmt.Sprintf("member%d@example.com", i)
			if err := fx.st.AddMember(OpsAccountID, addr, RoleMember, time.Now()); err != nil {
				t.Fatal(err)
			}
			tok = cliToken(t, fx.consoleFixture, addr, fmt.Sprintf("code-dbf-%d", i))
		}
		for j := 0; j < dbreads.OpenPerPerson; j++ {
			if len(held) == dbreads.OpenPerConsole {
				refusedTooMany(t, fx.openHeld(t, base, tok), "the seventeenth read on the console")
				continue
			}
			o := fx.admitWithin(t, base, tok)
			held = append(held, o)
		}
	}
	if len(held) != dbreads.OpenPerConsole {
		t.Fatalf("held %d reads", len(held))
	}

	// A restart under open reads: nothing to recover, nothing written, and
	// the new console's counts start empty.
	before, _ := fx.rel.List(context.Background(), dbreads.Prefix)
	srv2, err := NewServer(fx.srv.cfg, fx.st)
	if err != nil {
		t.Fatal(err)
	}
	srv2.dbf = fx.srv.dbf
	ts2 := httptest.NewServer(srv2)
	defer ts2.Close()
	after, _ := fx.rel.List(context.Background(), dbreads.Prefix)
	if !slices.Equal(before, after) {
		t.Fatalf("starting a console wrote under db-reads/: %v", after)
	}
	g := fx.openHeld(t, ts2.URL, fx.member)
	if g.id == "" {
		status, body := refusal(t, g.res)
		t.Fatalf("the restarted console refused a read: %d %s", status, body)
	}
	g.res.Body.Close()
	for _, o := range held {
		o.res.Body.Close()
	}
}

// slowStore holds every call whose key contains match until free closes,
// ignoring its context: the worst store, for plan review finding 12.
type slowStore struct {
	release.Store
	match string
	delay time.Duration // 0: until free closes
	free  chan struct{}
}

func (s *slowStore) hold(key string) {
	if !strings.Contains(key, s.match) {
		return
	}
	if s.delay > 0 {
		time.Sleep(s.delay)
		return
	}
	<-s.free
}

func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.hold(key)
	return s.Store.Get(ctx, key)
}

func (s *slowStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.hold(prefix)
	return s.Store.List(ctx, prefix)
}

// Plan review finding 12: a store that stops answering, in either phase,
// neither silences the stream nor holds it past the phase's deadline.
// Mutations: call the store on r.Context() without dbfCall (red: no wait
// line and no end until free closes); drop the pickup deadline context (red:
// same); drop the owner's deadline context (red: same in the owner case).
func TestDBReadStalledStoreKeepsTheStreamAndItsBound(t *testing.T) {
	for _, phase := range []string{"pickup", "owner"} {
		t.Run(phase, func(t *testing.T) {
			fx := newDBFFixture(t, true)
			fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.call = 10*time.Millisecond, 100*time.Millisecond, time.Minute
			fx.srv.dbf.pickup, fx.srv.dbf.grace = 1500*time.Millisecond, 0
			ss := &slowStore{Store: fx.rel, match: "/picked", free: make(chan struct{})}
			want := dbreads.EndedNotPicked
			if phase == "owner" {
				ss.match, want = "/result.json", dbreads.EndedNoResult
			}
			defer close(ss.free)
			fx.srv.cfg.Releases = ss
			o := fx.openHeld(t, fx.ts.URL, fx.member)
			defer o.res.Body.Close()
			if phase == "owner" {
				scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Second))
			}
			start, prev := time.Now(), time.Now()
			var last dbreads.Line
			for {
				line, err := o.br.ReadBytes('\n')
				if err != nil {
					break
				}
				if gap := time.Since(prev); gap > 500*time.Millisecond {
					t.Fatalf("a %s gap with the store stalled", gap)
				}
				prev = time.Now()
				json.Unmarshal(line, &last)
				if time.Since(start) > 5*time.Second {
					t.Fatal("the stalled store held the request past its deadline")
				}
			}
			if last.Ended != want {
				t.Fatalf("last line %+v, want %s", last, want)
			}
		})
	}
}

// Plan review finding 12: a relay slower than the time left ends at the
// request's bound, without the rest of the chunks and without the result,
// though both are in the bucket. issued_at is truncated to the second, so
// the bound is 2 to 3 s after the request; relaying all forty chunks takes 6.
// Mutation: relay without the owner's deadline, the chunk Gets on
// r.Context() and no ownCtx check after them (red: all forty are relayed
// and the request lasts 6 s).
// Not witnessed alone: the ownCtx check before the result line; dbfCall
// already cuts every call after the deadline, so that check guards only a
// line write that outlasts it, which no in-process client can force.
func TestDBReadSlowRelayEndsAtTheBound(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.bound, fx.srv.dbf.grace = 10*time.Millisecond, 3*time.Second, 0
	fx.srv.cfg.Releases = &slowStore{Store: fx.rel, match: "/out/", delay: 150 * time.Millisecond}
	start := time.Now()
	o := fx.openHeld(t, fx.ts.URL, fx.member)
	defer o.res.Body.Close()
	box := scriptFor(t, fx.rel, o.id)
	const n = 40
	for seq := int64(1); seq <= n; seq++ {
		box.chunk(seq, seq == n, "row\n")
	}
	box.result(n, strings.Repeat("row\n", n))
	box.picked(time.Now().Add(time.Minute))
	lines := readLines(t, o.br)
	chunks := 0
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a result relayed past the bound: %+v", l.Result)
		}
		if l.Chunk != 0 {
			chunks++
		}
	}
	if took := time.Since(start); took > 4500*time.Millisecond || chunks >= n {
		t.Fatalf("relayed %d chunks in %s", chunks, took)
	}
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
}

// scriptedBox writes a read's objects itself: a stand-in for a box, for the
// orderings T18 needs. It seals real chunks to the request's reply key.
type scriptedBox struct {
	t   *testing.T
	s   release.Store
	pub *ecdh.PublicKey
	id  string
}

func scriptFor(t *testing.T, s release.Store, id string) *scriptedBox {
	t.Helper()
	raw, _, err := s.Get(context.Background(), dbreads.RequestKey(id))
	if err != nil {
		t.Fatal(err)
	}
	var w struct{ Payload string }
	json.Unmarshal(raw, &w)
	payload, _ := base64.StdEncoding.DecodeString(w.Payload)
	var req dbreads.Request
	json.Unmarshal(payload, &req)
	pub, err := dbreads.ParseReplyKey(req.ReplyKey)
	if err != nil {
		t.Fatal(err)
	}
	return &scriptedBox{t: t, s: s, pub: pub, id: id}
}

func (b *scriptedBox) picked(deadline time.Time) {
	p, _ := json.Marshal(dbreads.Picked{Box: "uk-1", Nonce: "n", Deadline: deadline})
	b.s.Put(context.Background(), dbreads.PickedKey(b.id), p, "")
}

func (b *scriptedBox) chunk(seq int64, last bool, plain string) {
	sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamRows, last, []byte(plain))
	b.s.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "")
}

func (b *scriptedBox) result(last int64, rows string) {
	sum := sha256.Sum256([]byte(rows))
	sealed, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 9, Bytes: int64(len(rows)), SHA256: hex.EncodeToString(sum[:])})
	r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: last, Box: "uk-1", Sealed: sealed})
	b.s.Put(context.Background(), dbreads.ResultKey(b.id), r, "")
}

// dbReadHookStore runs once on the console's first listing of a read's
// chunks, after the listing is taken: a box committing between the console's
// listing and its next result read. It also records, in order, each result
// read and each chunk listing. (hookStore is temphook_test.go's: plan review
// finding 2.)
type dbReadHookStore struct {
	release.Store
	once sync.Once
	hook func()
	mu   sync.Mutex
	ops  []string
}

func (h *dbReadHookStore) record(op string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.ops = append(h.ops, op)
}

func (h *dbReadHookStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.HasSuffix(key, "/result.json") {
		h.record("result")
	}
	return h.Store.Get(ctx, key)
}

func (h *dbReadHookStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := h.Store.List(ctx, prefix)
	if strings.HasSuffix(prefix, "/out/") {
		h.record("list")
		h.once.Do(h.hook)
	}
	return keys, err
}

// T18 (codex 20): the box commits its last chunk and result.json between the
// console's listing and its result read, in both orders. The result line
// never comes before every chunk through last_seq, the read succeeds, and in
// every round the console reads result.json before it lists the chunks.
// Mutations, each red on its own (plan review finding 20): list out/ before
// reading result.json (the operation-order check), or write the result line
// without waiting for next > last_seq (the chunk-count check).
func TestDBReadResultWaitsForItsLastChunk(t *testing.T) {
	for _, resultFirst := range []bool{false, true} {
		fx := newDBFFixture(t, true)
		fx.srv.dbf.poll = 10 * time.Millisecond
		hs := &dbReadHookStore{Store: fx.rel}
		fx.srv.cfg.Releases = hs
		body, key := queryBody(t, "SELECT 1")
		res := fx.openRead(t, dbfHost, fx.member, body)
		br := bufio.NewReader(res.Body)
		first, _ := br.ReadBytes('\n')
		var adm dbreads.Line
		json.Unmarshal(first, &adm)
		box := scriptFor(t, fx.rel, adm.Admitted)
		hs.hook = func() {
			if resultFirst {
				box.result(2, "a\nb\n")
				box.chunk(2, true, "b\n")
			} else {
				box.chunk(2, true, "b\n")
				box.result(2, "a\nb\n")
			}
		}
		box.chunk(1, false, "a\n")
		box.picked(time.Now().Add(time.Minute))
		lines := append([]dbreads.Line{adm}, readLines(t, br)...)
		res.Body.Close()
		chunks := 0
		for _, l := range lines {
			if l.Chunk != 0 {
				chunks++
			}
			if l.Result != nil && chunks != 2 {
				t.Fatalf("result-first %v: the result line came after %d chunks", resultFirst, chunks)
			}
		}
		if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" {
			t.Fatalf("rows %q", rows)
		}
		hs.mu.Lock()
		ops := append([]string(nil), hs.ops...)
		hs.mu.Unlock()
		for i, op := range ops {
			if op == "list" && (i == 0 || ops[i-1] != "result") {
				t.Fatalf("result-first %v: a chunk listing not preceded by its round's result read: %v", resultFirst, ops)
			}
		}
	}
}

// dbfHideStore leaves chunk seq out of the first hide listings of a read's
// chunks: a listing that lags the box's writes.
type dbfHideStore struct {
	release.Store
	seq  int64
	mu   sync.Mutex
	hide int
}

func (h *dbfHideStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := h.Store.List(ctx, prefix)
	if !strings.HasSuffix(prefix, "/out/") {
		return keys, err
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.hide == 0 {
		return keys, err
	}
	h.hide--
	var out []string
	for _, k := range keys {
		if seq, ok := dbreads.ParseChunkKey(k, prefix); !ok || seq != h.seq {
			out = append(out, k)
		}
	}
	return out, err
}

// T18's completeness guard on its own (plan review finding 31): result.json
// says last_seq 2 from the first round, and the first three listings miss
// chunk 2. The result line waits until chunk 2 is relayed. Mutation: write
// the result line without waiting for next > res.LastSeq (red: the result
// comes after one chunk).
func TestDBReadResultWaitsForAChunkTheListingMissed(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll = 10 * time.Millisecond
	fx.srv.cfg.Releases = &dbfHideStore{Store: fx.rel, seq: 2, hide: 3}
	body, key := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.chunk(1, false, "a\n")
	box.chunk(2, true, "b\n")
	box.result(2, "a\nb\n")
	box.picked(time.Now().Add(time.Minute))
	lines := append([]dbreads.Line{adm}, readLines(t, br)...)
	chunks := 0
	for _, l := range lines {
		if l.Chunk != 0 {
			chunks++
		}
		if l.Result != nil && chunks != 2 {
			t.Fatalf("the result line came after %d chunks", chunks)
		}
	}
	if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" {
		t.Fatalf("rows %q", rows)
	}
}

// dbfHeldWriter is a ResponseWriter that holds the write of a line
// containing holdOn until until: a client that takes one line slowly. A
// stand-in for a network client, in a unit test.
type dbfHeldWriter struct {
	header http.Header
	holdOn string
	until  time.Time
	mu     sync.Mutex
	buf    bytes.Buffer
}

func (w *dbfHeldWriter) Header() http.Header { return w.header }
func (w *dbfHeldWriter) WriteHeader(int)     {}
func (w *dbfHeldWriter) Flush()              {}

func (w *dbfHeldWriter) Write(p []byte) (int, error) {
	if bytes.Contains(p, []byte(w.holdOn)) {
		time.Sleep(time.Until(w.until))
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.buf.Write(p)
}

// Plan review finding 35: the owner's deadline holds for the result line
// too. Every chunk and result.json are in the bucket, but the client takes
// the last chunk's line until past the deadline: the read ends no-result,
// and the result is not relayed. Mutation: drop the ownCtx check before the
// result line (red: the result is relayed).
func TestDBReadResultAfterTheDeadlineIsNotRelayed(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	key, err := dbreads.NewReplyKey()
	if err != nil {
		t.Fatal(err)
	}
	id, err := dbreads.MintID()
	if err != nil {
		t.Fatal(err)
	}
	box := &scriptedBox{t: t, s: fx.rel, pub: key.PublicKey(), id: id}
	deadline := time.Now().Add(800 * time.Millisecond)
	box.chunk(1, true, "a\n")
	box.result(1, "a\n")
	box.picked(deadline)
	w := &dbfHeldWriter{header: http.Header{}, holdOn: `"chunk":1,`, until: deadline.Add(200 * time.Millisecond)}
	r := httptest.NewRequest(http.MethodPost, "/", nil)
	fx.srv.streamDBRead(w, r, http.NewResponseController(w), dbReadAdmitted{id: id, region: "uk", issued: time.Now().UTC().Truncate(time.Second)})
	w.mu.Lock()
	lines := readLines(t, bytes.NewReader(w.buf.Bytes()))
	w.mu.Unlock()
	chunks := 0
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a result relayed past the owner's deadline: %+v", l.Result)
		}
		if l.Chunk != 0 {
			chunks++
		}
	}
	if last := lines[len(lines)-1]; chunks != 1 || last.Ended != dbreads.EndedNoResult {
		t.Fatalf("%d chunks, last line %+v", chunks, last)
	}
}

// dbfDeadlineListener records, for every read deadline the server sets on
// its connections, how far ahead it was set.
type dbfDeadlineListener struct {
	net.Listener
	mu    sync.Mutex
	ahead []time.Duration
}

type dbfDeadlineConn struct {
	net.Conn
	l *dbfDeadlineListener
}

func (l *dbfDeadlineListener) Accept() (net.Conn, error) {
	c, err := l.Listener.Accept()
	if err != nil {
		return nil, err
	}
	return dbfDeadlineConn{Conn: c, l: l}, nil
}

func (c dbfDeadlineConn) SetReadDeadline(t time.Time) error {
	if !t.IsZero() {
		c.l.mu.Lock()
		c.l.ahead = append(c.l.ahead, time.Until(t))
		c.l.mu.Unlock()
	}
	return c.Conn.SetReadDeadline(t)
}

// Plan review findings 13 and 38: admission reads under s.dbf.preHeader,
// and once admitted the read deadline moves out to the read's bound plus a
// minute. Observed on the connection itself: once a body has been read to
// its end, net/http's background read clears the read deadline on its own,
// so a stream that outlives ReadTimeout does not show the extension.
// Mutations: drop the extension (red: no deadline past 22 minutes); drop the
// preHeader deadline (red: none between 15 and 20 s; the server's own are
// 10 s and 5 minutes).
func TestDBReadSetsBothReadDeadlines(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.pickup = 10*time.Millisecond, 1500*time.Millisecond
	raw, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	ln := &dbfDeadlineListener{Listener: raw}
	hs := NewHTTPServer(fx.srv)
	go hs.Serve(ln)
	t.Cleanup(func() { hs.Close() })
	body, _ := queryBody(t, "SELECT 1")
	rawBody, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, "http://"+raw.Addr().String()+strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL), bytes.NewReader(rawBody))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+fx.member)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	io.Copy(io.Discard, res.Body)
	res.Body.Close()
	ln.mu.Lock()
	defer ln.mu.Unlock()
	var pre, extended bool
	for _, d := range ln.ahead {
		pre = pre || (d > 15*time.Second && d <= 20*time.Second)
		extended = extended || d > 22*time.Minute
	}
	if !pre || !extended {
		t.Fatalf("read deadlines set this far ahead: %v", ln.ahead)
	}
}

// Review Focus 3: a picked deadline far in the future does not hold the
// request past its own bound.
func TestDBReadRequestBoundClampsAForgedDeadline(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.bound = 10*time.Millisecond, 1500*time.Millisecond
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	scriptFor(t, fx.rel, adm.Admitted).picked(time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC))
	done := make(chan []dbreads.Line)
	go func() { done <- readLines(t, br) }()
	select {
	case lines := <-done:
		if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
			t.Fatalf("last line %+v", last)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the request outlived its bound")
	}
	res.Body.Close()
}

// Review Focus 4: a result.json that is not a result is never relayed.
func TestDBReadIgnoresAMalformedResult(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.picked(time.Now().Add(500 * time.Millisecond))
	fx.rel.Put(context.Background(), dbreads.ResultKey(adm.Admitted), []byte(`{"state":"maybe","last_seq":-1}`), "")
	lines := readLines(t, br)
	res.Body.Close()
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a malformed result was relayed: %+v", l.Result)
		}
	}
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
}

// strictStore fails the test on any delete or replacing write under
// db-reads/ (T16, the console's half).
type strictStore struct {
	release.Store
	t *testing.T
}

func (s strictStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if strings.HasPrefix(key, dbreads.Prefix) && ifMatch != "" {
		s.t.Errorf("a replacing write of %s", key)
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

func (s strictStore) Delete(ctx context.Context, key string) error {
	if strings.HasPrefix(key, dbreads.Prefix) {
		s.t.Errorf("a delete of %s", key)
	}
	return s.Store.Delete(ctx, key)
}

// T16 (codex 4, 13, 14): a box that writes after the request ended changes
// nothing any reader sees, and the console never deletes or replaces.
func TestDBReadLateWritesChangeNothing(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.cfg.Releases = strictStore{Store: fx.rel, t: t}
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.picked(time.Now().Add(200 * time.Millisecond))
	lines := readLines(t, br) // ends no-result at the deadline
	res.Body.Close()
	box.chunk(1, true, "late\n")
	box.result(1, "late\n")
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
	keys, _ := fx.rel.List(context.Background(), dbreads.Prefix+adm.Admitted+"/")
	for _, k := range keys {
		if strings.HasSuffix(k, "/cancel") {
			return
		}
	}
	t.Fatalf("no cancel among %v", keys)
}
```

Add imports `"crypto/sha256"`, `"encoding/base64"`, `"encoding/hex"`, `"slices"`, `lsfile "github.com/benbjohnson/litestream/file"`, `dbdoortest`, `dbreadstest`.

- [ ] **Step 2: Run them to verify they fail**

Run: `go test ./internal/console/ -run 'TestDBReadStreams|TestDBReadEnds|TestDBReadLines|TestDBReadAdmissionBounds|TestDBReadStalledStore|TestDBReadSlowRelay|TestDBReadResultWaits|TestDBReadResultAfterTheDeadline|TestDBReadSetsBothReadDeadlines|TestDBReadRequestBound|TestDBReadIgnores|TestDBReadLateWrites' -count=1`
Expected: FAIL: the stub writes only `admitted`.

- [ ] **Step 3: Write the stream**

Replace the stub in `internal/console/apidbforward.go` with:

```go
// errDBClientGone is a line the client did not take within s.dbf.stall.
var errDBClientGone = errors.New("the client went or stalled")

// dbfEndSlack is how far past the request's bound the last line may still be
// written: room for the ended line, and no more.
const dbfEndSlack = 10 * time.Second

// dbLineWriter writes one line at a time, each under a fresh write deadline
// and flushed under the same deadline (spec §5.5 step 6, codex 19). The
// deadline never passes until, the request's bound plus dbfEndSlack (plan
// review finding 12). A write or flush error ends the request as a stalled
// client.
type dbLineWriter struct {
	w     http.ResponseWriter
	rc    *http.ResponseController
	stall time.Duration
	until time.Time
	last  time.Time
}

func (l *dbLineWriter) write(line dbreads.Line) error {
	b, err := json.Marshal(line)
	if err != nil {
		return err
	}
	dl := time.Now().Add(l.stall)
	if dl.After(l.until) {
		dl = l.until
	}
	if err := l.rc.SetWriteDeadline(dl); err != nil && !errors.Is(err, http.ErrNotSupported) {
		return err
	}
	if _, err := l.w.Write(append(b, '\n')); err != nil {
		return err
	}
	if err := l.rc.Flush(); err != nil {
		return err
	}
	l.last = time.Now()
	return nil
}

// tick waits one poll and writes a wait line when nothing has been written
// for every. It answers ctx's error when ctx ends first, and errDBClientGone
// when the line cannot be written.
func (l *dbLineWriter) tick(ctx context.Context, phase string, poll, every time.Duration) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(poll):
	}
	if time.Since(l.last) >= every && l.write(dbreads.Line{Wait: phase}) != nil {
		return errDBClientGone
	}
	return nil
}

// dbReadRelay is one admitted read's relay, with what its last log line
// counts. (dbStream is apidb.go's streaming function: plan review finding 25.)
type dbReadRelay struct {
	s      *Server
	lw     *dbLineWriter
	chunks int
	bytes  int
}

// dbfCall runs one store call under s.dbf.call and the phase's context, and
// keeps writing wait lines while it blocks, so a stalled store neither holds
// the request past its phase nor silences the stream (plan review finding
// 12). A store that ignores its context is left behind: its goroutine ends
// when the call does, into a buffered channel nobody reads.
func dbfCall[T any](st *dbReadRelay, ctx context.Context, phase string, op func(context.Context) (T, error)) (T, error) {
	var zero T
	cctx, cancel := context.WithTimeout(ctx, st.s.dbf.call)
	defer cancel()
	type outcome struct {
		v   T
		err error
	}
	done := make(chan outcome, 1)
	go func() {
		v, err := op(cctx)
		done <- outcome{v, err}
	}()
	t := time.NewTicker(st.s.dbf.poll)
	defer t.Stop()
	for {
		select {
		case o := <-done:
			return o.v, o.err
		case <-cctx.Done():
			return zero, cctx.Err()
		case <-t.C:
			if time.Since(st.lw.last) >= st.s.dbf.wait && st.lw.write(dbreads.Line{Wait: phase}) != nil {
				return zero, errDBClientGone
			}
		}
	}
}

// readDBResult is result.json, or nil while there is none, and nil for one
// that is not a result in the vocabulary (Review Focus 4).
func (s *Server) readDBResult(ctx context.Context, id string) *dbreads.Result {
	raw, _, err := s.cfg.Releases.Get(ctx, dbreads.ResultKey(id))
	if err != nil {
		return nil
	}
	var res dbreads.Result
	if json.Unmarshal(raw, &res) != nil || res.Valid() != nil {
		return nil
	}
	return &res
}

// streamDBRead is spec §5.5 steps 6 and 7 inside the admitted request:
// admitted at once, picked once, then about once a second result.json first
// and the chunk listing second (codex 20; the run follow door's order), each
// chunk in sequence with no gaps, a wait line whenever s.dbf.wait passes
// with nothing else, and the result last, only once every chunk through its
// last_seq has been written. Each phase runs on its own deadline context:
// pickup on issued_at plus s.dbf.pickup, the owner's on its deadline plus
// s.dbf.grace, never later than s.dbf.bound after issued_at. It ends without
// a result, and writes cancel, when either deadline passes, a result
// included, or when the client goes or stalls. Its last log line is the
// relay's own measure (plan review finding 21).
func (s *Server) streamDBRead(w http.ResponseWriter, r *http.Request, rc *http.ResponseController, a dbReadAdmitted) {
	ctx := r.Context()
	bound := a.issued.Add(s.dbf.bound)
	w.Header().Set("Content-Type", "application/x-ndjson")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(http.StatusOK)
	st := &dbReadRelay{s: s, lw: &dbLineWriter{w: w, rc: rc, stall: s.dbf.stall, until: bound.Add(dbfEndSlack)}}
	start, why := time.Now(), "client gone"
	defer func() {
		log.Printf("console: db read %s: relayed %d chunks (%d bytes) in %s, ended %s",
			a.id, st.chunks, st.bytes, time.Since(start).Round(time.Millisecond), why)
	}()
	end := func(ended string) {
		if ended != "" {
			why = ended
			st.lw.write(dbreads.Line{Ended: ended})
		}
		s.cancelDBRead(a.id)
	}
	// stop ends a phase: a client that went or stalled gets no line, a phase
	// whose deadline passed gets its ended word.
	stop := func(err error, ended string) {
		if errors.Is(err, errDBClientGone) || ctx.Err() != nil {
			end("")
			return
		}
		end(ended)
	}
	if st.lw.write(dbreads.Line{Admitted: a.id}) != nil {
		end("")
		return
	}
	store := s.cfg.Releases

	pickCtx, cancelPick := context.WithDeadline(ctx, a.issued.Add(s.dbf.pickup))
	defer cancelPick()
	var picked dbreads.Picked
	for {
		raw, err := dbfCall(st, pickCtx, dbreads.WaitPickup, func(c context.Context) ([]byte, error) {
			b, _, err := store.Get(c, dbreads.PickedKey(a.id))
			return b, err
		})
		if err == nil && json.Unmarshal(raw, &picked) == nil {
			break
		}
		if pickCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNotPicked)
			return
		}
		if err := st.lw.tick(pickCtx, dbreads.WaitPickup, s.dbf.poll, s.dbf.wait); err != nil {
			stop(err, dbreads.EndedNotPicked)
			return
		}
	}
	if st.lw.write(dbreads.Line{Picked: picked.Box}) != nil {
		end("")
		return
	}
	deadline := picked.Deadline.Add(s.dbf.grace)
	if deadline.After(bound) {
		deadline = bound
	}
	ownCtx, cancelOwn := context.WithDeadline(ctx, deadline)
	defer cancelOwn()
	next := int64(1)
	for {
		res, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) (*dbreads.Result, error) {
			return s.readDBResult(c, a.id), nil
		}) // first
		if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNoResult)
			return
		}
		keys, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) ([]string, error) {
			return store.List(c, dbreads.OutPrefix(a.id))
		})
		if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNoResult)
			return
		}
		if err == nil {
			listed := map[int64]bool{}
			for _, k := range keys {
				if seq, ok := dbreads.ParseChunkKey(k, dbreads.OutPrefix(a.id)); ok {
					listed[seq] = true
				}
			}
			for listed[next] {
				seq := next
				sealed, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) ([]byte, error) {
					b, _, err := store.Get(c, dbreads.ChunkKey(a.id, seq))
					return b, err
				})
				if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
					stop(err, dbreads.EndedNoResult)
					return
				}
				if err != nil {
					break
				}
				if st.lw.write(dbreads.Line{Chunk: seq, Data: sealed}) != nil {
					end("")
					return
				}
				st.chunks++
				st.bytes += len(sealed)
				next++
			}
		}
		if res != nil && next > res.LastSeq {
			// The bound holds for a result too: one that is complete only
			// after the deadline is not relayed (plan review finding 12).
			if ownCtx.Err() != nil {
				stop(nil, dbreads.EndedNoResult)
				return
			}
			why = res.State
			st.lw.write(dbreads.Line{Result: res})
			return
		}
		if err := st.lw.tick(ownCtx, dbreads.WaitReading, s.dbf.poll, s.dbf.wait); err != nil {
			stop(err, dbreads.EndedNoResult)
			return
		}
	}
}
```

A result whose `last_seq` is below a chunk already written cannot occur from an honest box (the box writes the result after its last chunk). If it does, `next > res.LastSeq` still holds and the CLI's own checks refuse the answer (DBF-13).

- [ ] **Step 4: Run the tests**

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

- [ ] **Step 5: Commit**

```bash
git add internal/console
git commit -m "console: the reads route streams the read: result before listing, flushed lines, cancel on every end"
```

---

### Task 8: The CLI's forwarded query

Waits for Task 1 (the CLI rows approved).

**Files:**
- Create: `dbforward.go`
- Create: `dbforward_test.go`
- Modify: `db.go:187-195` (`dbRestore` hands a forward marker back), `db.go:277-290` (`dbPull`), `db.go:374-382` (`dbQuery`)
- Modify: `approved_db_forward_copy_test.go` (DBF-3 to DBF-8, DBF-11 to DBF-13, DBF-15, DBF-16, DBF-32)

**Interfaces:**
- Consumes: `dbdoor.DoorError.Forward`, `dbdoor.Forward` (Task 2); `dbreads.NewReplyKey`, `EncodeReplyKey`, `ReadBody`, `Line`, `OpenChunk`, `OpenSummary`, `MaxLineBytes`, `CLIBound`, `StreamRows`, the state and code constants; `consoleClient.doStream`, `apiPath`, `apiErrorFromBody`, `dbQueryStatus`, `dbRestoreError`, `humanBytes`, `checkQuerySQL`, `stdout` (existing CLI code).
- Produces: `errDBForward` (an error type carrying `dbdoor.Forward`); `dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error`; `readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region string, status io.Writer) (dbForwardAnswer, error)`; `type dbfChunkBound`, `newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound`, `(*dbfChunkBound).admit(sealed []byte) bool`; `openDBForward(id string, key *ecdh.PrivateKey, sealed [][]byte, res dbreads.Result, host, region string) (dbForwardAnswer, error)`; `type dbForwardAnswer struct{ rows []byte; summary dbreads.Summary; capped bool }`; the `dbf*` CLI constants.

- [ ] **Step 1: Write the failing tests**

Create `dbforward_test.go`. It feeds `readDBForward` crafted line streams, sealed with real keys: unit tests of the CLI's half (Task 10 runs the binary).

```go
package main

import (
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

const fwdID = "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"

// stream builds a reads-route body: admitted, picked, the chunks, then end.
type stream struct {
	t   *testing.T
	pub *ecdh.PublicKey
	buf bytes.Buffer
}

func newStream(t *testing.T, key *ecdh.PrivateKey) *stream {
	s := &stream{t: t, pub: key.PublicKey()}
	s.line(dbreads.Line{Admitted: fwdID})
	s.line(dbreads.Line{Picked: "uk-1"})
	return s
}

func (s *stream) line(l dbreads.Line) {
	b, _ := json.Marshal(l)
	s.buf.Write(append(b, '\n'))
}

func (s *stream) chunk(seq int64, last bool, plain []byte) {
	sealed, err := dbreads.SealChunk(s.pub, fwdID, seq, dbreads.StreamRows, last, plain)
	if err != nil {
		s.t.Fatal(err)
	}
	s.line(dbreads.Line{Chunk: seq, Data: sealed})
}

func (s *stream) result(state, code string, last int64, sum dbreads.Summary) {
	sealed, _ := dbreads.SealSummary(s.pub, fwdID, sum)
	s.line(dbreads.Line{Result: &dbreads.Result{State: state, Code: code, LastSeq: last, Sealed: sealed}})
}

func summaryOf(rows string) dbreads.Summary {
	h := sha256.Sum256([]byte(rows))
	return dbreads.Summary{TXID: 7, At: time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC), Bytes: int64(len(rows)), SHA256: hex.EncodeToString(h[:])}
}

func readFwd(t *testing.T, s *stream, key *ecdh.PrivateKey) (dbForwardAnswer, string, error) {
	t.Helper()
	var status bytes.Buffer
	ans, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", &status)
	return ans, status.String(), err
}

func TestForwardedQueryPrintsOnlyACheckedAnswer(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	s.chunk(2, true, []byte("30\n"))
	s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
	ans, status, err := readFwd(t, s, key)
	if err != nil || string(ans.rows) != "n\n30\n" || ans.summary.TXID != 7 {
		t.Fatalf("%q %v", ans.rows, err)
	}
	if status != fmt.Sprintf(dbfReading, "shop.example.com", "uk")+"\n" {
		t.Fatalf("status %q", status)
	}
}

// T11's unit half (codex 6): a dropped or bad last chunk and a summary that
// does not add up each print nothing.
func TestForwardedQueryRefusesAnAnswerThatDoesNotCheckOut(t *testing.T) {
	t.Parallel()
	unchecked := fmt.Sprintf(dbfUnchecked, "uk")
	for name, build := range map[string]func(s *stream){
		"the last chunk dropped": func(s *stream) {
			s.chunk(1, false, []byte("n\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
		},
		"the last chunk not sealed as last": func(s *stream) {
			s.chunk(1, false, []byte("n\n"))
			s.chunk(2, false, []byte("30\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
		},
		"a hash mismatch": func(s *stream) {
			s.chunk(1, true, []byte("n\n31\n"))
			s.result(dbreads.StateDone, "", 1, summaryOf("n\n30\n"))
		},
		"a chunk out of order": func(s *stream) {
			s.chunk(2, true, []byte("30\n"))
			s.result(dbreads.StateDone, "", 1, summaryOf("30\n"))
		},
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		build(s)
		ans, _, err := readFwd(t, s, key)
		if err == nil || err.Error() != unchecked || ans.rows != nil {
			t.Errorf("%s: %q, %v", name, ans.rows, err)
		}
	}
}

func TestForwardedQueryEndsWithoutAResult(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	if _, _, err := readFwd(t, s, key); err == nil || err.Error() != fmt.Sprintf(dbfUnfinished, "shop.example.com", "uk") {
		t.Fatalf("no result: %v", err)
	}
	key2, _ := dbreads.NewReplyKey()
	s2 := &stream{t: t, pub: key2.PublicKey()}
	s2.line(dbreads.Line{Admitted: fwdID})
	s2.line(dbreads.Line{Wait: dbreads.WaitPickup})
	s2.line(dbreads.Line{Ended: dbreads.EndedNotPicked})
	if _, _, err := readFwd(t, s2, key2); err == nil || err.Error() != fmt.Sprintf(dbfNotPicked, "uk") {
		t.Fatalf("not picked: %v", err)
	}
}

func TestForwardedQueryMapsEveryCode(t *testing.T) {
	t.Parallel()
	host := "shop.example.com"
	for code, want := range map[string]string{
		dbreads.CodeTooLarge:          fmt.Sprintf(dbfTooLarge, host, "3.0GB", "2.0GB", "uk"),
		dbreads.CodeNoRoom:            fmt.Sprintf(dbfNoRoom, "uk", host),
		dbreads.CodeTimeout:           dbfTimeout,
		dbreads.CodeSQLError:          "no such table: nope",
		dbreads.CodeNothingReplicated: dbRestoreError(host, dbdoorErrNothing()).Error(),
		dbreads.CodeChangedTwice:      dbRestoreError(host, errVanished()).Error(),
		dbreads.CodeInterrupted:       fmt.Sprintf(dbfUnfinished, host, "uk"),
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		sum := dbreads.Summary{Restored: 3 << 30, Cap: 2 << 30, Error: "no such table: nope"}
		s.result(dbreads.StateFailed, code, 0, sum)
		if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want {
			t.Errorf("%s: %v, want %q", code, err, want)
		}
	}
	for code, want := range map[string]string{
		dbreads.CodeRecordChanged: fmt.Sprintf(dbfChanged, host),
		dbreads.CodeMoving:        fmt.Sprintf(dbfMovingCLI, host),
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		s.result(dbreads.StateRefused, code, 0, dbreads.Summary{})
		if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want {
			t.Errorf("%s: %v", code, err)
		}
	}
}

func TestForwardedQueryOutputCapPrintsTheRows(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, true, []byte("cut"))
	s.result(dbreads.StateDone, dbreads.CodeOutputCap, 1, summaryOf("cut"))
	ans, _, err := readFwd(t, s, key)
	if err != nil || !ans.capped || string(ans.rows) != "cut" {
		t.Fatalf("%+v %v", ans, err)
	}
}

// Review Focus 2: a full-size chunk line, about 350 KiB, is read.
func TestForwardedQueryReadsAFullSizeChunkLine(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	rows := strings.Repeat("x", dbreads.ChunkBytes)
	s := newStream(t, key)
	s.chunk(1, false, []byte(rows))
	s.chunk(2, true, nil)
	s.result(dbreads.StateDone, "", 2, summaryOf(rows))
	ans, _, err := readFwd(t, s, key)
	if err != nil || len(ans.rows) != dbreads.ChunkBytes {
		t.Fatalf("%d bytes, %v", len(ans.rows), err)
	}
}

// Review Focus 1: Ctrl-C mid-stream is "stopped", with nothing printed.
func TestForwardedQueryCtrlCPrintsNothing(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	var status bytes.Buffer
	ans, err := readDBForward(ctx, &cutReader{r: &s.buf}, key, "shop.example.com", "uk", &status)
	if err == nil || err.Error() != "stopped" || ans.rows != nil {
		t.Fatalf("%q %v", ans.rows, err)
	}
}

// Plan review finding 37: what the CLI holds before a result is bounded by
// count as well as bytes, and a malformed chunk is refused before it is
// kept. Each stream is otherwise one the CLI would accept, so each case is
// red under its own mutation of dbfChunkBound.admit: drop the size floor
// ("a chunk shorter than its seal" then ends DBF-12, the stream having no
// result, instead of DBF-13 at once); drop the empty-chunk rule or the
// count bound (the stream then opens and prints).
func TestForwardedQueryBoundsChunkCountAndSize(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	unchecked := fmt.Sprintf(dbfUnchecked, "uk")
	many := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes).maxChunks + 1
	for name, write := range map[string]func(s *stream){
		"a chunk shorter than its seal": func(s *stream) {
			s.line(dbreads.Line{Chunk: 1, Data: []byte("short")})
		},
		"a chunk after an empty one": func(s *stream) {
			s.chunk(1, false, nil)
			s.chunk(2, true, []byte("n\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n"))
		},
		"more chunks than a read can make": func(s *stream) {
			for seq := int64(1); seq <= many; seq++ {
				s.chunk(seq, seq == many, []byte("x"))
			}
			s.result(dbreads.StateDone, "", many, summaryOf(strings.Repeat("x", int(many))))
		},
	} {
		s := newStream(t, key)
		write(s)
		var status bytes.Buffer
		if _, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", &status); err == nil || err.Error() != unchecked {
			t.Errorf("%s: %v", name, err)
		}
	}
}

// Plan review finding 36: SQL that JSON would rewrite is refused before
// anything is sent. Mutation: drop the check (red: the request goes out).
func TestForwardedQueryRefusesSQLThatIsNotUTF8(t *testing.T) {
	t.Parallel()
	var sent atomic.Bool
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		sent.Store(true)
		http.Error(w, "no", http.StatusTeapot)
	}))
	defer ts.Close()
	target := dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"},
		app: "shop", host: "shop.example.com"}
	for _, q := range []string{"SELECT '\xff'", "SELECT '\uFFFD'"} {
		err := dbQueryForward(context.Background(), target, q, dbdoor.Forward{Region: "uk", Ops: []string{dbreads.OpQuery}})
		if err == nil || err.Error() != fmt.Sprintf(dbfNotUTF8, "uk") {
			t.Errorf("%q: %v", q, err)
		}
	}
	if sent.Load() {
		t.Fatal("a query JSON would rewrite was sent")
	}
}

// cutReader ends like a connection the context closed.
type cutReader struct{ r *bytes.Buffer }

func (c *cutReader) Read(p []byte) (int, error) {
	if c.r.Len() == 0 {
		return 0, context.Canceled
	}
	return c.r.Read(p)
}

// The errors the in-region restore answers with, for the mapping test.
func dbdoorErrNothing() error { return dbdoor.ErrNothingReplicated }

func errVanished() error { return os.ErrNotExist }
```

- [ ] **Step 2: Run them to verify they fail**

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

- [ ] **Step 3: Write `dbforward.go`**

```go
// dbforward.go is the CLI's half of a forwarded `carlos db query` (spec
// docs/superpowers/specs/2026-10-22-db-region-forward-design.md §5.5): when
// the db door answers with the forward marker, the CLI makes an ephemeral
// key, holds one request to the console's reads route open, buffers every
// sealed chunk, and prints nothing until a done result's summary checks out
// against the bytes (codex 6).
package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"time"

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

// The forwarded read's CLI sentences (copy record).
const (
	dbfNotPicked  = "no server in region %s picked up the read within 60 seconds. try again in a few minutes."                    // DBF-3
	dbfTooLarge   = "%s's database is %s once restored. carlos db can query at most %s in region %s."                              // DBF-4
	dbfNoRoom     = "the server in region %s does not have room to restore %s's database right now. try again later."             // DBF-5
	dbfOutputCap  = "the result was cut off at 64 MB. narrow the query with WHERE or LIMIT and run it again."                       // DBF-6
	dbfTimeout    = "the query ran for 5 minutes and was stopped. narrow it and run it again."                                     // DBF-7
	dbfNoPull     = "%s keeps its data in region %s. carlos db pull cannot fetch a copy from there yet. carlos db query can read it." // DBF-8
	dbfReading    = "reading a copy of %s's database in region %s."                                                                // DBF-11
	dbfUnfinished = "the read of %s's database in region %s did not finish, so nothing was printed. try again."                    // DBF-12
	dbfUnchecked  = "the answer from region %s did not check out, so carlos db showed none of it. try again."                      // DBF-13
	dbfChanged    = "%s changed while the read was waiting. try again."                                                            // DBF-15
	dbfMovingCLI  = "%s is moving between regions. read its database once the move has finished."                                 // DBF-16
	dbfNotUTF8    = "this query cannot be sent to region %s, because it holds bytes that are not UTF-8 text or the character U+FFFD. fix the query and run it again." // DBF-32
)

// errDBForward is the db door's forward marker, handed back by dbRestore so
// a verb can switch to the forwarded read.
type errDBForward struct {
	door *dbdoor.DoorError
}

func (e *errDBForward) Error() string { return e.door.Error() }

type dbForwardAnswer struct {
	rows    []byte
	summary dbreads.Summary
	capped  bool
}

// dbQueryForward is spec §5.5 steps 2 and 8 for a query.
func dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error {
	// json.Marshal would send U+FFFD for bytes that are not UTF-8, and the
	// box would run SQL nobody typed; the console refuses U+FFFD for the
	// same reason (plan review finding 36). First, so no other check can
	// read the bytes differently.
	if !utf8.ValidString(sqlText) || strings.ContainsRune(sqlText, utf8.RuneError) {
		return fmt.Errorf(dbfNotUTF8, fwd.Region)
	}
	if err := checkQuerySQL(sqlText); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound)
	defer cancel()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		return err
	}
	at := ""
	if !t.at.IsZero() {
		at = t.at.UTC().Format(time.RFC3339Nano)
	}
	body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	if err != nil {
		return 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 errors.New("stopped")
		}
		return 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 apiErrorFromBody(res.StatusCode, raw)
	}
	ans, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, os.Stderr)
	if err != nil {
		return err
	}
	fmt.Fprint(os.Stderr, dbQueryStatus(t.host, &dbdoor.Restored{TXID: ans.summary.TXID, At: ans.summary.At}, t.at))
	if _, err := stdout.Write(ans.rows); err != nil {
		return err
	}
	if ans.capped {
		return errors.New(dbfOutputCap)
	}
	return nil
}

// dbfChunkBound is what a reader keeps before a result arrives (plan review
// finding 37). Every chunk is at least a sealed empty one; only the last
// may be empty, so nothing may follow an empty chunk; there are at most as
// many chunks as a read can make (full ones for the payload, plus one per
// flush interval over the owner's deadline, plus the last); and their
// plaintext, ciphertext less each seal, stays within the payload cap.
type dbfChunkBound struct {
	maxBytes, maxChunks int64
	n, plain            int64
	sawEmpty            bool
}

func newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound {
	return &dbfChunkBound{maxBytes: maxBytes,
		maxChunks: (maxBytes+chunkBytes-1)/chunkBytes + int64(dbreads.OwnerDeadline/dbreads.ChunkEvery) + 1}
}

// admit says whether a chunk may be kept; false is DBF-13.
func (b *dbfChunkBound) admit(sealed []byte) bool {
	size := int64(len(sealed)) - int64(dbreads.SealOverhead)
	if size < 0 || b.sawEmpty || b.n >= b.maxChunks || b.plain+size > b.maxBytes {
		return false
	}
	b.n++
	b.plain += size
	b.sawEmpty = size == 0
	return true
}

// readDBForward reads the reads route's lines (spec §5.5 steps 6 to 8). It
// says DBF-11 once admitted, buffers every chunk in sequence, and opens and
// checks the answer only when the result arrives.
func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region string, status io.Writer) (dbForwardAnswer, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	sc := bufio.NewScanner(body)
	sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
	var id, ended string
	var sealed [][]byte
	bound := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)
	for sc.Scan() {
		var l dbreads.Line
		if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
			return dbForwardAnswer{}, unchecked
		}
		switch {
		case l.Admitted != "":
			if id != "" || !dbreads.ValidID(l.Admitted) {
				return dbForwardAnswer{}, unchecked
			}
			id = l.Admitted
			fmt.Fprintln(status, fmt.Sprintf(dbfReading, host, region))
		case l.Chunk != 0:
			if id == "" || l.Chunk != int64(len(sealed))+1 || !bound.admit(l.Data) {
				return dbForwardAnswer{}, unchecked
			}
			sealed = append(sealed, l.Data)
		case l.Result != nil:
			if id == "" {
				return dbForwardAnswer{}, unchecked
			}
			return openDBForward(id, key, sealed, *l.Result, host, region)
		case l.Ended != "":
			ended = l.Ended
		}
	}
	if errors.Is(ctx.Err(), context.Canceled) {
		return dbForwardAnswer{}, errors.New("stopped")
	}
	if ended == dbreads.EndedNotPicked {
		return dbForwardAnswer{}, fmt.Errorf(dbfNotPicked, region)
	}
	return dbForwardAnswer{}, fmt.Errorf(dbfUnfinished, host, region)
}

// openDBForward opens a result: a done one's every chunk and its summary,
// checked against each other before anything is returned; any other state
// mapped to its sentence (spec §5.6, §10).
func openDBForward(id string, key *ecdh.PrivateKey, sealed [][]byte, res dbreads.Result, host, region string) (dbForwardAnswer, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	if res.Valid() != nil {
		return dbForwardAnswer{}, unchecked
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		return dbForwardAnswer{}, unchecked
	}
	switch res.State {
	case dbreads.StateDone:
		if res.LastSeq != int64(len(sealed)) {
			return dbForwardAnswer{}, unchecked
		}
		var rows bytes.Buffer
		for i, c := range sealed {
			seq := int64(i + 1)
			plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamRows, seq == res.LastSeq, c)
			if err != nil {
				return dbForwardAnswer{}, unchecked
			}
			rows.Write(plain)
		}
		sum := sha256.Sum256(rows.Bytes())
		if int64(rows.Len()) != s.Bytes || hex.EncodeToString(sum[:]) != s.SHA256 {
			return dbForwardAnswer{}, unchecked
		}
		return dbForwardAnswer{rows: rows.Bytes(), summary: s, capped: res.Code == dbreads.CodeOutputCap}, nil
	case dbreads.StateRefused:
		if res.Code == dbreads.CodeMoving {
			return dbForwardAnswer{}, fmt.Errorf(dbfMovingCLI, host)
		}
		return dbForwardAnswer{}, fmt.Errorf(dbfChanged, host)
	case dbreads.StateFailed:
		switch res.Code {
		case dbreads.CodeNothingReplicated:
			return dbForwardAnswer{}, dbRestoreError(host, dbdoor.ErrNothingReplicated)
		case dbreads.CodeHistoryGone:
			if !s.Oldest.IsZero() {
				return dbForwardAnswer{}, dbRestoreError(host, &dbdoor.BeforeHistoryError{Oldest: s.Oldest})
			}
		case dbreads.CodeCheckFailed:
			return dbForwardAnswer{}, dbRestoreError(host, fmt.Errorf("%w: %s", dbdoor.ErrCheckFailed, s.Error))
		case dbreads.CodeChangedTwice:
			return dbForwardAnswer{}, dbRestoreError(host, os.ErrNotExist)
		case dbreads.CodeTooLarge:
			return dbForwardAnswer{}, fmt.Errorf(dbfTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region)
		case dbreads.CodeNoRoom:
			return dbForwardAnswer{}, fmt.Errorf(dbfNoRoom, region, host)
		case dbreads.CodeSQLError:
			return dbForwardAnswer{}, errors.New(s.Error)
		case dbreads.CodeTimeout:
			return dbForwardAnswer{}, errors.New(dbfTimeout)
		}
	}
	return dbForwardAnswer{}, fmt.Errorf(dbfUnfinished, host, region)
}
```

`fmt.Fprintln(status, fmt.Sprintf(dbfReading, ...))` keeps the format a single `Sprintf` the copy gate can read; `go vet` accepts it.

In `db.go`, make `dbRestore` hand the marker back:

```go
func dbRestore(ctx context.Context, t dbTarget) (*dbdoor.Restored, error) {
	r, err := dbdoor.RestoreAsOf(ctx, t.sess.dbClient(t.acct, t.app, t.host), "", t.at)
	if err != nil {
		var door *dbdoor.DoorError
		if errors.As(err, &door) && door.Forward != nil {
			return nil, &errDBForward{door: door}
		}
		return nil, dbRestoreError(t.host, err)
	}
	return r, nil
}
```

In `dbQuery`, before the existing body continues:

```go
	r, err := dbRestore(ctx, t)
	var fwd *errDBForward
	if errors.As(err, &fwd) {
		if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpQuery) {
			return errors.New(fwd.door.Error())
		}
		return dbQueryForward(ctx, t, sqlText, *fwd.door.Forward)
	}
```

and in `dbPull`:

```go
	r, err := dbRestore(ctx, t)
	var fwd *errDBForward
	if errors.As(err, &fwd) {
		return fmt.Errorf(dbfNoPull, t.host, fwd.door.Forward.Region)
	}
```

Add `"slices"` and the `dbreads` import to `db.go`. Add the CLI ids (DBF-3 to DBF-8, DBF-11, DBF-12, DBF-13, DBF-15, DBF-16, DBF-32) to `dbForwardCopyBuilt`, and `"unicode/utf8"` to `dbforward.go`'s imports (and `"strings"` if it is not there).

- [ ] **Step 4: Run the tests and the gate**

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

- [ ] **Step 5: Commit**

```bash
git add dbforward.go dbforward_test.go db.go approved_db_forward_copy_test.go
git commit -m "carlos db query: the forwarded read, buffered and checked before anything prints"
```

---

### Task 9: The CLI page (DBF-30)

Waits for Task 1 (DBF-30 approved).

**Files:**
- Modify: `docs/site/cli.md:539-542` (the `carlos db` paragraph that starts "If the app has more than one instance")
- Modify: `approved_db_forward_copy_test.go` (DBF-30)

- [ ] **Step 1: Add DBF-30 to the built list and run the gate**

Run: `go test . -run 'TestDBForwardBuiltCopyIsApplied' -count=1`
Expected: FAIL: the CLI page lacks the text.

- [ ] **Step 2: Replace the paragraph**

Replace the four lines from "If the app has more than one instance, name one with `--instance`." to "and the command says which." with the approved DBF-30 text, each paragraph on one line, a blank line between them (the gate matches the record's `\n\n`). The text as drafted in spec §10:

```markdown
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.

When 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. The rows are encrypted for your terminal on the way, so nothing in between can read them, and nothing is printed until all of them have arrived and checked out. There, the 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. `carlos db pull` cannot fetch a copy from another region yet.
```

Write Paul's approved text if it differs.

- [ ] **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**

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

---

### Task 10: Acceptance through the real CLI: T6, T7, T8, T11

**Files:**
- Create: `dbforward_e2e_test.go`

**Interfaces:**
- Consumes: `buildCarlos`, `useTempCredentials`, `clearDeploymentEnv`, `carlosTestEnv`, `e2eDeviceLogin`, `loadCredentials`, `credential`, `replicaS3` (`db_e2e_test.go`), `dbdoortest.Sync`, `console.Open`, `console.NewServer`, `console.NewHTTPServer`, `console.OpsAccountID`, `console.RoleOwner`, `edge.New`, `registry.Open`, `registry.Route`, `dbreadstest.Start`, `dbreads.*`, `dbdoor.RestoreLocal`, `dbdoor.NewClient`, `dbdoor.Base`, `lease.ReadReplicaClient`.
- Produces: the harness `dbfE2E` and `newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E`; `(*dbfE2E).run(t, args...) (stdout, stderr string, err error)`; `(*dbfE2E).betweenTheSyncs(t) string`; `(*dbfE2E).doer(host string) dbdoor.Doer`; `(*dbfE2E).scriptBox(t, script func(*e2eScript)) (stop func())`; `e2eScript` with `picked(deadline)`, `chunk(seq, last, plain)`, `result(last, rows)`; `throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap)`; `dbfTap`, `newDBFTap(t, target string, connect bool) *dbfTap`, `(*dbfTap).longestGap(start, end time.Time) time.Duration`; `checkProgressAtTheCLI(t, e, tap, atLeast time.Duration)`; `slowRestore(d time.Duration) dbreads.WorkerOptions`; and the tests below.

- [ ] **Step 1: Write the harness**

Create `dbforward_e2e_test.go`:

```go
// Acceptance for the forwarded `carlos db query` through the real CLI
// binary and the real console server: T6 (parity with the in-region door,
// through the fake S3 the db door's own e2e uses), T7 (the long request
// through the real edge and the console's real http.Server), T8 (mixed
// versions) and T11 (nothing printed for an answer that does not check out).
// The box is plan 1's real agent and worker code joined in process
// (dbreadstest): not acceptance for the units, which plan 1's root-runner
// tests are. A scripted box (e2eScript) is a stand-in and says so.
package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"

	"github.com/carlosframework/platform/internal/console"
	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/dbreads/dbreadstest"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

const (
	dbfE2EHere = "shop.example.com" // in the console's region, ie
	dbfE2EFar  = "far.example.com"  // in uk, with the same database bytes
)

type dbfE2EOptions struct {
	forward bool
	// serve serves the console's handler and returns the URL the CLI uses
	// and any extra environment it needs; nil is the httptest server.
	serve func(t *testing.T, h http.Handler) (consoleURL string, env []string)
	// worker is the in-process box's worker options (a slow restore for T7).
	worker dbreads.WorkerOptions
	// noBox leaves the uk box out; the test scripts its own.
	noBox bool
}

type dbfE2E struct {
	bin, creds, url, tok, root string
	env                        []string
	store                      release.Store
	pub                        ed25519.PublicKey
	priv                       ed25519.PrivateKey
}

// newDBFE2E is a console in ie over a DirStore and the fake S3: shop in ie
// and far in uk with byte-identical replicas (the same files with the same
// modification times, so the fake S3 lists the same LastModified for both),
// a capable live uk box, and a signed-in CLI.
func newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E {
	t.Helper()
	creds := useTempCredentials(t)
	clearDeploymentEnv(t)
	tmp := t.TempDir()
	e := &dbfE2E{bin: buildCarlos(t), creds: creds, store: release.NewDirStore(filepath.Join(tmp, "store")), root: filepath.Join(tmp, "data")}
	var err error
	if e.pub, e.priv, err = ed25519.GenerateKey(rand.Reader); err != nil {
		t.Fatal(err)
	}
	here := filepath.Join(e.root, "instances", dbfE2EHere)
	dbfE2EFixture(t, here)
	copyTreeKeepingTimes(t, here, filepath.Join(e.root, "instances", dbfE2EFar))
	s3 := httptest.NewServer(&replicaS3{t: t, bucket: "data", root: e.root})
	t.Cleanup(s3.Close)
	t.Setenv("CARLOS_S3_ENDPOINT", s3.URL)
	t.Setenv("AWS_ACCESS_KEY_ID", "test")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")

	cst, err := console.Open(filepath.Join(tmp, "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { cst.Close() })
	var srv *console.Server
	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) })
	direct := httptest.NewServer(h)
	t.Cleanup(direct.Close)
	e.url = direct.URL
	if o.serve != nil {
		e.url, e.env = o.serve(t, h)
	}
	srv, err = console.NewServer(console.Config{
		Origin: direct.URL, Keymail: "http://keymail.invalid", Registry: filepath.Join(tmp, "registry.db"),
		Releases: e.store, ReleasePub: e.pub, Key: e.priv, AppScheme: "http",
		Region: "ie", HibernateBucket: "data", HibernateRegion: "us-east-1", HibernatePrefix: "instances",
		DBForward: o.forward,
	}, cst)
	if err != nil {
		t.Fatal(err)
	}
	const addr = "member@example.com"
	if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, err := cst.CreateAppIn(console.OpsAccountID, "shop", addr, time.Now()); err != nil {
		t.Fatal(err)
	}
	created := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
	for host, region := range map[string]string{dbfE2EHere: "ie", dbfE2EFar: "uk"} {
		yes := true
		body, _ := json.Marshal(provision.Record{Host: host, App: "shop", Account: release.OpsAccount, Region: region, Hibernate: &yes, CreatedAt: created})
		if _, err := e.store.Put(context.Background(), provision.RecordKey(release.OpsAccount, "shop", host), body, ""); err != nil {
			t.Fatal(err)
		}
	}
	edges, _ := json.Marshal(map[string]any{"v": 1, "box": "uk-1", "eip": "203.0.113.10", "placement": "uk",
		"capabilities": []string{dbreads.Capability}, "hosts": map[string]any{}})
	e.store.Put(context.Background(), "control/edges/uk-1.json", edges, "")
	status, _ := json.Marshal(map[string]any{"generated_at": time.Now().UTC(), "label": "uk-1"})
	e.store.Put(context.Background(), "status/boxes/uk-1.json", status, "")
	token, expiresAt := e2eDeviceLogin(t, direct.URL, cst, addr, "e2e@laptop")
	e.tok = token
	cf, err := loadCredentials()
	if err != nil {
		t.Fatal(err)
	}
	cf.put(credential{Console: e.url, Address: addr, Token: token, ExpiresAt: expiresAt})
	if err := cf.save(); err != nil {
		t.Fatal(err)
	}
	if !o.noBox {
		box := dbreadstest.Start(t, dbreadstest.Options{Store: e.store, Region: "uk", Box: "uk-1", Pub: e.pub,
			HibernatePrefix: "instances", Worker: o.worker,
			OpenReplica: func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error) {
				return lease.ReadReplicaClient(ctx, "data", prefix, "us-east-1")
			}})
		ctx, cancel := context.WithCancel(context.Background())
		done := make(chan struct{})
		go func() {
			defer close(done)
			for ctx.Err() == nil {
				box.Agent.Pass(ctx)
				time.Sleep(50 * time.Millisecond)
			}
		}()
		t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() })
	}
	return e
}

// dbfE2EFixture is T6's database (NULL, a tab and a newline in one value, a
// blob, a time) in two syncs a whole second apart.
func dbfE2EFixture(t *testing.T, dir string) {
	t.Helper()
	src := filepath.Join(t.TempDir(), "shop.db")
	db, err := sql.Open("sqlite", src)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	for _, q := range []string{
		`PRAGMA journal_mode=WAL`,
		`CREATE TABLE orders (id INTEGER PRIMARY KEY, note TEXT, data BLOB, at DATETIME)`,
	} {
		if _, err := db.Exec(q); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`INSERT INTO orders (note, data, at) VALUES (?, ?, ?), (NULL, NULL, NULL)`,
		"a\ttab\nline", []byte{0, 0xff}, time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC)); err != nil {
		t.Fatal(err)
	}
	dbdoortest.Sync(t, src, dir)
	time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(1100 * time.Millisecond)))
	if _, err := db.Exec(`INSERT INTO orders (note) VALUES ('later')`); err != nil {
		t.Fatal(err)
	}
	dbdoortest.Sync(t, src, dir)
}

func copyTreeKeepingTimes(t *testing.T, from, to string) {
	t.Helper()
	err := filepath.Walk(from, func(p string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(from, p)
		dst := filepath.Join(to, rel)
		if info.IsDir() {
			return os.MkdirAll(dst, 0o755)
		}
		b, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		if err := os.WriteFile(dst, b, 0o644); err != nil {
			return err
		}
		return os.Chtimes(dst, info.ModTime(), info.ModTime())
	})
	if err != nil {
		t.Fatal(err)
	}
}

func (e *dbfE2E) run(t *testing.T, args ...string) (string, string, error) {
	t.Helper()
	cmd := exec.Command(e.bin, args...)
	cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...)
	cmd.Dir = t.TempDir()
	var stdout, stderr bytes.Buffer
	cmd.Stdout, cmd.Stderr = &stdout, &stderr
	err := cmd.Run()
	return stdout.String(), stderr.String(), err
}

// betweenTheSyncs is an --at that holds the first sync and not the second:
// half a second into the second after the first L0 file was written. The
// second sync is at least 1.1 s after the first (dbfE2EFixture), so its
// listed second is never before this time's.
func (e *dbfE2E) betweenTheSyncs(t *testing.T) string {
	t.Helper()
	ents, err := os.ReadDir(filepath.Join(e.root, "instances", dbfE2EHere, "ltx", "0"))
	if err != nil || len(ents) < 2 {
		t.Fatalf("the fixture has %d L0 files: %v", len(ents), err)
	}
	var first time.Time
	for _, ent := range ents {
		info, err := ent.Info()
		if err != nil {
			t.Fatal(err)
		}
		if first.IsZero() || info.ModTime().Before(first) {
			first = info.ModTime()
		}
	}
	return first.Truncate(time.Second).Add(1500 * time.Millisecond).UTC().Format(time.RFC3339Nano)
}

// doer is an old CLI's view of the db door: GETs with the saved token.
func (e *dbfE2E) doer(host string) dbdoor.Doer {
	return func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, e.url+dbdoor.Base(release.OpsAccount, "shop", host)+tail, nil)
		if err != nil {
			return nil, err
		}
		for k, vs := range h {
			req.Header[k] = vs
		}
		req.Header.Set("Authorization", "Bearer "+e.tok)
		return http.DefaultClient.Do(req)
	}
}

// e2eScript writes one read's objects itself: a stand-in for a box, sealing
// real chunks and a real summary to the request's reply key.
type e2eScript struct {
	store release.Store
	id    string
	pub   *ecdh.PublicKey
}

func (b *e2eScript) picked(deadline time.Time) {
	p, _ := json.Marshal(dbreads.Picked{Box: "uk-1", Nonce: "scripted", Deadline: deadline})
	b.store.Put(context.Background(), dbreads.PickedKey(b.id), p, "")
}

func (b *e2eScript) chunk(seq int64, last bool, plain string) {
	sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamRows, last, []byte(plain))
	b.store.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "")
}

func (b *e2eScript) result(last int64, rows string) {
	sum := sha256.Sum256([]byte(rows))
	sealed, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 9, Bytes: int64(len(rows)), SHA256: hex.EncodeToString(sum[:])})
	r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: last, Box: "uk-1", Sealed: sealed})
	b.store.Put(context.Background(), dbreads.ResultKey(b.id), r, "")
}

// scriptBox runs script once for each new request in uk's queue, after
// verifying it as a box would.
func (e *dbfE2E) scriptBox(t *testing.T, script func(*e2eScript)) func() {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		seen := map[string]bool{}
		for ctx.Err() == nil {
			keys, _ := e.store.List(ctx, dbreads.QueueRegionPrefix("uk"))
			for _, k := range keys {
				q, ok := dbreads.ParseQueueKey(k)
				if !ok || seen[q.ID] {
					continue
				}
				seen[q.ID] = true
				raw, _, err := e.store.Get(ctx, dbreads.RequestKey(q.ID))
				if err != nil {
					continue
				}
				req, err := dbreads.Verify(e.pub, raw, q.ID, "uk", time.Now())
				if err != nil {
					t.Errorf("the scripted box refused %s: %v", q.ID, err)
					continue
				}
				pub, _ := dbreads.ParseReplyKey(req.ReplyKey)
				go script(&e2eScript{store: e.store, id: q.ID, pub: pub})
			}
			time.Sleep(50 * time.Millisecond)
		}
	}()
	return func() { cancel(); <-done }
}

// throughTheEdge serves the console's handler with console.NewHTTPServer on
// 127.0.0.1, its ReadTimeout set to readTimeout, behind a real edge route
// for console.test. The CLI reaches it by that name through HTTP_PROXY,
// which stands in for DNS.
func throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) {
	return func(t *testing.T, h http.Handler) (string, []string) {
		ln, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			t.Fatal(err)
		}
		hs := console.NewHTTPServer(h)
		hs.ReadTimeout = readTimeout
		go hs.Serve(ln)
		t.Cleanup(func() { hs.Close() })
		reg, err := registry.Open(filepath.Join(t.TempDir(), "edge.db"))
		if err != nil {
			t.Fatal(err)
		}
		if err := reg.Upsert(registry.Route{Host: "console.test", Kind: "service", Addr: ln.Addr().String(), Status: "ready"}); err != nil {
			t.Fatal(err)
		}
		ed, err := edge.New(reg, nil)
		if err != nil {
			t.Fatal(err)
		}
		if err := ed.Reload(); err != nil {
			t.Fatal(err)
		}
		es := httptest.NewServer(ed)
		t.Cleanup(es.Close)
		*edgeURL = es.URL
		// The CLI's proxy is a tap in front of the edge, so the test sees
		// what reaches the CLI's own socket (plan review finding 34).
		tp := newDBFTap(t, es.Listener.Addr().String(), false)
		*tap = tp
		return "http://console.test", []string{"HTTP_PROXY=http://" + tp.addr, "NO_PROXY="}
	}
}

// dbfTap stands between the CLI and the first edge, forwarding bytes both
// ways and noting when each read from the edge's side was passed to the
// CLI's socket: what the CLI process received, and when. With connect it is
// an HTTPS proxy (CONNECT), which is how the CLI reaches a TLS edge by name
// here; without, a plain forwarder for HTTP_PROXY.
type dbfTap struct {
	addr string
	mu   sync.Mutex
	down []time.Time
}

func newDBFTap(t *testing.T, target string, connect bool) *dbfTap {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { ln.Close() })
	tp := &dbfTap{addr: ln.Addr().String()}
	go func() {
		for {
			c, err := ln.Accept()
			if err != nil {
				return
			}
			go tp.serve(c, target, connect)
		}
	}()
	return tp
}

func (tp *dbfTap) serve(c net.Conn, target string, connect bool) {
	defer c.Close()
	var in io.Reader = c
	if connect {
		br := bufio.NewReader(c)
		req, err := http.ReadRequest(br)
		if err != nil || req.Method != http.MethodConnect {
			return
		}
		if _, err := io.WriteString(c, "HTTP/1.1 200 Connection established\r\n\r\n"); err != nil {
			return
		}
		in = br
	}
	up, err := net.Dial("tcp", target)
	if err != nil {
		return
	}
	defer up.Close()
	go func() {
		io.Copy(up, in)
		up.(*net.TCPConn).CloseWrite()
	}()
	buf := make([]byte, 32<<10)
	for {
		n, err := up.Read(buf)
		if n > 0 {
			tp.mu.Lock()
			tp.down = append(tp.down, time.Now())
			tp.mu.Unlock()
			if _, werr := c.Write(buf[:n]); werr != nil {
				return
			}
		}
		if err != nil {
			return
		}
	}
}

// longestGap is the longest silence at the CLI's socket between start and
// end, both of which count as arrivals.
func (tp *dbfTap) longestGap(start, end time.Time) time.Duration {
	tp.mu.Lock()
	defer tp.mu.Unlock()
	prev, longest := start, time.Duration(0)
	for _, at := range append(append([]time.Time(nil), tp.down...), end) {
		if at.Before(start) || at.After(end) {
			continue
		}
		if d := at.Sub(prev); d > longest {
			longest = d
		}
		prev = at
	}
	return longest
}

// checkProgressAtTheCLI is T7's witness at the CLI process itself (plan
// review finding 34): it runs the real CLI and needs DBF-11 on its stderr
// within a second of the process starting (stricter than the spec's "of the
// request": the process starts before it sends anything), no silence over
// 16 s (15 s and a poll) at its socket, the read lasting at least atLeast, and
// its rows. atLeast is the quiet restore's length.
func checkProgressAtTheCLI(t *testing.T, e *dbfE2E, tap *dbfTap, atLeast time.Duration) {
	t.Helper()
	cmd := exec.Command(e.bin, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
	cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...)
	cmd.Dir = t.TempDir()
	var out bytes.Buffer
	cmd.Stdout = &out
	pipe, err := cmd.StderrPipe()
	if err != nil {
		t.Fatal(err)
	}
	start := time.Now()
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	var stderr []string
	var reading time.Time
	sc := bufio.NewScanner(pipe)
	for sc.Scan() {
		if reading.IsZero() && strings.Contains(sc.Text(), fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) {
			reading = time.Now()
		}
		stderr = append(stderr, sc.Text())
	}
	err = cmd.Wait()
	end := time.Now()
	if err != nil || out.String() != "n\n3\n" {
		t.Fatalf("%q %v\n%s", out.String(), err, strings.Join(stderr, "\n"))
	}
	if reading.IsZero() || reading.Sub(start) > time.Second {
		t.Fatalf("DBF-11 after %s: %q", reading.Sub(start), stderr)
	}
	if gap := tap.longestGap(start, end); gap > 16*time.Second {
		t.Fatalf("%s of silence at the CLI's socket", gap)
	}
	if took := end.Sub(start); took < atLeast {
		t.Fatalf("the read took %s, under %s: it did not hold the request", took, atLeast)
	}
}

// slowRestore holds the box's restore for d before restoring for real.
func slowRestore(d time.Duration) dbreads.WorkerOptions {
	return dbreads.WorkerOptions{Restore: func(ctx context.Context, c litestream.ReplicaClient, out string, target uint64) error {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(d):
		}
		return dbdoor.RestoreLocal(ctx, c, out, target)
	}}
}
```

- [ ] **Step 2: Write the tests**

Append to `dbforward_e2e_test.go`:

```go
// T6: the forwarded query and the in-region query print byte-identical
// stdout, for the newest copy and as of a time between the two syncs.
func TestDBForwardParityThroughTheCLI(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true})
	const q = "SELECT id, note, data, at FROM orders ORDER BY id"
	for _, at := range []string{"", e.betweenTheSyncs(t)} {
		args := []string{"db", "query", "--app", "shop", "--sql", q}
		if at != "" {
			args = append(args, "--at", at)
		}
		here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere)...)
		if err != nil {
			t.Fatalf("in region: %v\n%s", err, hereErr)
		}
		far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar)...)
		if err != nil {
			t.Fatalf("forwarded: %v\n%s", err, farErr)
		}
		if here != far || here == "" {
			t.Fatalf("at %q: in region %q, forwarded %q", at, here, far)
		}
		if !strings.Contains(farErr, fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) {
			t.Fatalf("no DBF-11 on stderr: %s", farErr)
		}
		if at != "" && strings.Contains(far, "later") {
			t.Fatalf("as of %s the copy holds the second sync: %q", at, far)
		}
	}
}

// T7, scaled: the console's real http.Server with ReadTimeout 2 s instead of
// 5 minutes, behind the real edge, and a read held 20 s by a slow restore.
// The connection outlives ReadTimeout tenfold, admitted arrives within a
// second, a line arrives at least every 15 s (plus a poll), and the real CLI
// gets its rows through the same path.
func TestDBForwardLongRequestThroughTheEdge(t *testing.T) {
	var edgeURL string
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughTheEdge(2*time.Second, &edgeURL, &tap), worker: slowRestore(20 * time.Second)})
	key, _ := dbreads.NewReplyKey()
	body, _ := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: "SELECT count(*) AS n FROM orders", ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	req, _ := http.NewRequest(http.MethodPost, edgeURL+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfE2EFar+"/reads", bytes.NewReader(body))
	req.Host = "console.test"
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+e.tok)
	start := time.Now()
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	prev, first, sawResult := start, true, false
	for {
		line, err := br.ReadBytes('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			t.Fatal(err)
		}
		gap := time.Since(prev)
		if first && gap > time.Second {
			t.Fatalf("admitted took %s", gap)
		}
		if gap > 16*time.Second {
			t.Fatalf("%s with no line", gap)
		}
		prev, first = time.Now(), false
		sawResult = sawResult || bytes.Contains(line, []byte(`"result"`))
	}
	if !sawResult || time.Since(start) < 20*time.Second {
		t.Fatalf("result %v after %s", sawResult, time.Since(start))
	}
	// The same through the real CLI process, observed at its own stderr and
	// socket (plan review finding 34).
	checkProgressAtTheCLI(t, e, tap, 20*time.Second)
}

// T7, unscaled: a read that holds the request for 21 minutes through the
// real edge and the console's unmodified http.Server (5-minute ReadTimeout),
// inside the CLI's 23-minute context. The box is scripted (a stand-in):
// picked at 1 minute 50 seconds with a 20-minute deadline, one chunk at 10
// minutes, the last and the result at 21. It waits; it does not compute.
// Run once before handoff with CARLOS_DBF_LONG=1.
func TestDBForwardTwentyOneMinuteRead(t *testing.T) {
	if os.Getenv("CARLOS_DBF_LONG") != "1" {
		t.Skip("set CARLOS_DBF_LONG=1 to run the unscaled 21-minute read, once, before handoff")
	}
	var edgeURL string
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: throughTheEdge(5*time.Minute, &edgeURL, &tap)})
	stop := e.scriptBox(t, func(b *e2eScript) {
		time.Sleep(110 * time.Second)
		b.picked(time.Now().Add(20 * time.Minute))
		time.Sleep(8 * time.Minute)
		b.chunk(1, false, "n\n")
		time.Sleep(11*time.Minute + 10*time.Second)
		b.chunk(2, true, "3\n")
		b.result(2, "n\n3\n")
	})
	defer stop()
	start := time.Now()
	out, stderr, err := e.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
	if err != nil || out != "n\n3\n" || time.Since(start) < 21*time.Minute {
		t.Fatalf("after %s: %q %v\n%s", time.Since(start), out, err, stderr)
	}
}

// T8: the rows of spec §9.2 the CLI sees.
func TestDBForwardMixedVersions(t *testing.T) {
	off := newDBFE2E(t, dbfE2EOptions{forward: false, noBox: true})
	// A new CLI and a console with forwarding off (or an old console): the
	// residency sentence as it is, and nothing on stdout.
	out, stderr, err := off.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 1")
	if err == nil || out != "" || !strings.Contains(stderr, "far.example.com keeps its data in region uk. This console can only read data kept in region ie.") {
		t.Fatalf("new CLI, forwarding off: %q %q %v", out, stderr, err)
	}
	on := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	// An old CLI and a console with forwarding on: every CLI from before
	// this plan runs dbRestoreError on a door refusal, which prints the
	// door's sentence. That sentence is DBF-14.
	_, derr := dbdoor.NewClient(on.doer(dbfE2EFar)).LTXFiles(context.Background(), 9, 0, false)
	want := "far.example.com keeps its data in region uk. this version of carlos cannot read it from there. run carlos update, then try again."
	if got := dbRestoreError(dbfE2EFar, derr).Error(); got != want {
		t.Fatalf("old CLI, new console: %q", got)
	}
}

// T11 (codex 6): the real CLI prints nothing for a dropped last chunk, a
// summary hash mismatch, or a response that ends without a result. The box
// is scripted: a stand-in that writes picked, chunks and result itself.
func TestDBForwardPrintsNothingForABadAnswer(t *testing.T) {
	for name, c := range map[string]struct {
		script func(b *e2eScript)
		want   string
	}{
		// The bucket lost the real last chunk: the result names chunk 1,
		// which was sealed as not the last.
		"the last chunk dropped": {func(b *e2eScript) {
			b.picked(time.Now().Add(time.Minute))
			b.chunk(1, false, "n\n")
			b.result(1, "n\n")
		}, "did not check out"},
		"a summary hash mismatch": {func(b *e2eScript) {
			b.picked(time.Now().Add(time.Minute))
			b.chunk(1, true, "n\n31\n")
			b.result(1, "n\n30\n")
		}, "did not check out"},
		// The console waits the owner's deadline plus its 1-minute grace.
		"no result": {func(b *e2eScript) {
			b.picked(time.Now().Add(2 * time.Second))
			b.chunk(1, false, "n\n")
		}, "did not finish, so nothing was printed"},
	} {
		t.Run(name, func(t *testing.T) {
			e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
			stop := e.scriptBox(t, c.script)
			defer stop()
			out, stderr, err := e.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
			if err == nil || out != "" || !strings.Contains(stderr, c.want) {
				t.Fatalf("%q %q %v", out, stderr, err)
			}
		})
	}
}

// Plan review finding 36, through the binary: a query holding a byte that
// is not UTF-8 is refused with DBF-32 before anything is sent, so no altered
// query reaches the bucket. (An argv may carry any bytes on Linux, where
// this runs.)
func TestDBForwardRefusesSQLThatIsNotUTF8(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	out, stderr, err := e.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 'caf\xe9'")
	if err == nil || out != "" || !strings.Contains(stderr, fmt.Sprintf(dbfNotUTF8, "uk")) {
		t.Fatalf("%q %q %v", out, stderr, err)
	}
	if keys, _ := e.store.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a query JSON would rewrite reached the bucket: %v", keys)
	}
}
```

- [ ] **Step 3: Run them**

Run (sandbox off; they build the CLI binary):
```bash
go test . -run 'TestDBForwardParityThroughTheCLI|TestDBForwardMixedVersions|TestDBForwardPrintsNothingForABadAnswer|TestDBForwardRefusesSQLThatIsNotUTF8|TestDBForwardLongRequestThroughTheEdge' -count=1 -v
```
Expected: PASS; `TestDBForwardTwentyOneMinuteRead` skips. If the edge answers the CLI with 404, the request reached it with a host other than `console.test`: print `r.Host` in a wrapper around `ed` to see what Go's proxy client sent.

- [ ] **Step 4: Commit**

```bash
git add dbforward_e2e_test.go
git commit -m "db forward: T6, T7, T8 and T11 through the real CLI binary and console"
```

---

### Task 11: T7's backhaul half, through the real CLI and console

**Files:**
- Create: `internal/edge/dbforward_backhaul_helper_test.go`
- Modify: `dbforward_e2e_test.go`

**Interfaces:**
- Consumes (package `edge` test helpers in `backhaul_e2e_test.go`): `newE2ECA`, `(*e2eCA).publish`, the `e2eCA.root` field, `newBackhaulEdge`, `(*backhaulEdge).tlsAddr`, `.mirror`, `.homes`, `refreshHomeMap`, `MintMeshKey`, `putJSON`, `e2eQuietLogs`, the seams `backhaulPort` and `backhaulRootCAs`. From Task 10: `newDBFE2E`, `slowRestore`, `dbfTap`, `newDBFTap`, `checkProgressAtTheCLI`.
- Produces: `TestDBForwardBackhaulPairHelper` (a helper process, skipped unless `DBF_BACKHAUL_HELPER=1`); `throughBackhaul(readTimeout time.Duration, tap **dbfTap)`; `TestDBForwardThroughTheBackhaulHop`.

Spec T7 needs the real CLI and the real console through the backhaul hop (plan review finding 34). The console cannot be imported by package `edge`'s tests (the console imports the edge, a cycle), and the backhaul harness is package `edge` test code that nothing else can import. So the acceptance test lives in the root package with the real CLI and console, and the two edges run in package `edge`'s own test binary as a helper process, built by the root test with `go test -c`.

- [ ] **Step 1: Write the helper**

Create `internal/edge/dbforward_backhaul_helper_test.go`:

```go
package edge

import (
	"context"
	"encoding/pem"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"testing"

	"github.com/carlosframework/platform/internal/release"
)

const dbfBackhaulHost = "console.backhaul.test"

// TestDBForwardBackhaulPairHelper is not a test of its own: the root
// package's TestDBForwardThroughTheBackhaulHop builds this package's test
// binary and runs this as a helper process. It stands up two real edges
// over one bucket, as TestBackhaulE2EAcrossTwoEdges does: annie, home of
// console.backhaul.test, routed to DBF_BACKHAUL_UPSTREAM (the real console
// the parent serves), and betty, which serves no host and so backhauls every
// request for it to annie. It writes the test CA to DBF_BACKHAUL_DIR/ca.pem,
// prints betty's TLS address on a line starting "DBF-BACKHAUL ", and serves
// until its stdin closes.
func TestDBForwardBackhaulPairHelper(t *testing.T) {
	if os.Getenv("DBF_BACKHAUL_HELPER") != "1" {
		t.Skip("a helper process for the root package's TestDBForwardThroughTheBackhaulHop")
	}
	upstream, dir := os.Getenv("DBF_BACKHAUL_UPSTREAM"), os.Getenv("DBF_BACKHAUL_DIR")
	e2eQuietLogs(t)
	bucket := t.TempDir()
	seed := release.NewDirStore(bucket)
	if err := MintMeshKey(context.Background(), seed); err != nil {
		t.Fatal(err)
	}
	ca := newE2ECA(t)
	ca.publish(t, seed, dbfBackhaulHost)
	home := newBackhaulEdge(t, bucket, "annie", ca, nil, map[string]string{dbfBackhaulHost: upstream})
	sender := newBackhaulEdge(t, bucket, "betty", ca, nil, nil)
	eip, port, err := net.SplitHostPort(home.tlsAddr)
	if err != nil {
		t.Fatal(err)
	}
	oldPort, oldCAs := backhaulPort, backhaulRootCAs
	backhaulPort, backhaulRootCAs = port, ca.pool
	t.Cleanup(func() { backhaulPort, backhaulRootCAs = oldPort, oldCAs })
	putJSON(t, seed, "control/edges/annie.json", fmt.Sprintf(`{"v":1,"box":"annie","eip":%q,"hosts":{%q:{"role":"home"}}}`, eip, dbfBackhaulHost))
	bEIP, _, err := net.SplitHostPort(sender.tlsAddr)
	if err != nil {
		t.Fatal(err)
	}
	putJSON(t, seed, "control/edges/betty.json", fmt.Sprintf(`{"v":1,"box":"betty","eip":%q,"hosts":{}}`, bEIP))
	for _, be := range []*backhaulEdge{home, sender} {
		if err := be.mirror.Refresh(context.Background()); err != nil {
			t.Fatal(err)
		}
		refreshHomeMap(t, be.homes)
	}
	caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.root.Raw})
	if err := os.WriteFile(filepath.Join(dir, "ca.pem"), caPEM, 0o600); err != nil {
		t.Fatal(err)
	}
	fmt.Println("DBF-BACKHAUL " + sender.tlsAddr)
	io.Copy(io.Discard, os.Stdin)
}
```

Run (sandbox off): `go test ./internal/edge/ -run 'TestDBForwardBackhaulPairHelper' -count=1`
Expected: `ok` with the test skipped.

- [ ] **Step 2: Write the acceptance test**

Append to `dbforward_e2e_test.go`:

```go
// throughBackhaul serves the console's handler with console.NewHTTPServer
// (its ReadTimeout set to readTimeout) on 127.0.0.1, behind two real edges
// that package edge's test binary runs as a helper process: the CLI reaches
// betty, which backhauls to annie, the console's home. The CLI finds betty
// through a CONNECT proxy (a dbfTap, standing in for DNS) and trusts the
// edges' test CA through SSL_CERT_FILE.
func throughBackhaul(readTimeout time.Duration, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) {
	return func(t *testing.T, h http.Handler) (string, []string) {
		ln, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			t.Fatal(err)
		}
		hs := console.NewHTTPServer(h)
		hs.ReadTimeout = readTimeout
		go hs.Serve(ln)
		t.Cleanup(func() { hs.Close() })
		dir := t.TempDir()
		bin := filepath.Join(dir, "edge.test")
		if out, err := exec.Command("go", "test", "-c", "-o", bin, "./internal/edge").CombinedOutput(); err != nil {
			t.Fatalf("build the edge test binary: %v\n%s", err, out)
		}
		cmd := exec.Command(bin, "-test.run", "^TestDBForwardBackhaulPairHelper$", "-test.v")
		cmd.Env = append(os.Environ(), "DBF_BACKHAUL_HELPER=1", "DBF_BACKHAUL_UPSTREAM="+ln.Addr().String(), "DBF_BACKHAUL_DIR="+dir)
		stdin, err := cmd.StdinPipe()
		if err != nil {
			t.Fatal(err)
		}
		pr, pw := io.Pipe()
		cmd.Stdout = pw
		var helperErr bytes.Buffer
		cmd.Stderr = &helperErr
		if err := cmd.Start(); err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() {
			stdin.Close()
			cmd.Wait()
			pw.Close()
			if t.Failed() {
				t.Logf("backhaul helper stderr:\n%s", helperErr.String())
			}
		})
		lines := bufio.NewScanner(pr)
		sender := ""
		for lines.Scan() {
			if addr, ok := strings.CutPrefix(lines.Text(), "DBF-BACKHAUL "); ok {
				sender = addr
				break
			}
		}
		if sender == "" {
			t.Fatalf("the backhaul helper never came up:\n%s", helperErr.String())
		}
		go func() {
			for lines.Scan() {
			}
		}()
		tp := newDBFTap(t, sender, true)
		*tap = tp
		return "https://console.backhaul.test", []string{"HTTPS_PROXY=http://" + tp.addr, "NO_PROXY=", "SSL_CERT_FILE=" + filepath.Join(dir, "ca.pem")}
	}
}

// T7, the backhaul half, end to end (plan review finding 34): the real CLI
// binary reaches the real console through two real edges, the second hop a
// backhaul with its 30 s header bound, during a 40 s quiet restore. At the
// CLI's own stderr and socket: DBF-11 within a second, no silence over 16 s,
// and its rows after more than the hop's 30 s. The console's real
// http.Server runs with ReadTimeout 2 s, as in
// TestDBForwardLongRequestThroughTheEdge.
// Mutations: drop the stream's flush (red: the wait lines sit in the
// console's buffer and the CLI's socket is silent); stream nothing before
// the box's first chunk (red: no headers inside the hop's 30 s).
func TestDBForwardThroughTheBackhaulHop(t *testing.T) {
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughBackhaul(2*time.Second, &tap), worker: slowRestore(40 * time.Second)})
	checkProgressAtTheCLI(t, e, tap, 40*time.Second)
}
```

- [ ] **Step 3: Run it**

Run (sandbox off; it builds the CLI and the edge test binary): `go test . -run 'TestDBForwardThroughTheBackhaulHop' -count=1 -v`
Expected: PASS after about 45 s. If the helper never comes up, its stderr is in the log. If the CLI gets 404 from betty instead of a backhaul, betty's home map did not see annie's edges file: compare with `newBackhaulScene`'s order (mesh key, certificates, edges files, then each edge's mirror and home map) and follow it. If the CLI fails its TLS handshake, check that `SSL_CERT_FILE` reached it (`carlosTestEnv` appends the harness's extras last).

- [ ] **Step 4: Commit**

```bash
git add internal/edge/dbforward_backhaul_helper_test.go dbforward_e2e_test.go
git commit -m "db forward: T7 through the backhaul hop with the real CLI and console"
```

---

### Task 12: T1 and T16 against MinIO

**Files:**
- Create: `internal/console/dbforward_minio_test.go`
- Modify: `internal/console/server_test.go:49-93` (`newConsoleFixtureOn`)
- Modify: `.amadan/ci.d/30-minio-contract` (buckets, users, the run and its PASS check)

**Interfaces:**
- Consumes: `dbfFixture`, `queryBody`, `openRead`, `readLines`, `openAnswer`, `putEdgesFile`, `putBoxStatus`, `runBox`, `strictStore` (Tasks 4 to 7); `newConsoleFixtureKeyed`'s body; `release.S3Store`; `lss3.NewReplicaClient`; `dbdoortest.Replicate`; `dbreadstest.Start`.
- Produces: `newConsoleFixtureOn(t *testing.T, withKey bool, rel release.Store) *consoleFixture`; `uploadReplica(t, from *lsfile.ReplicaClient, to *lss3.ReplicaClient)`; `dbfGateStore`; `TestDBForwardMinio`.

- [ ] **Step 1: Let the console fixture run over another store**

In `internal/console/server_test.go`, rename `newConsoleFixtureKeyed`'s body into `newConsoleFixtureOn(t *testing.T, withKey bool, rel release.Store) *consoleFixture`, and right after `regPath, relStore, priv, pub := buildFixtures(t)` add:

```go
	// A test that needs a real S3 store (TestDBForwardMinio) passes its own;
	// the fixture's seeded releases are then absent, which those tests do not
	// read.
	if rel != nil {
		relStore = rel
	}
```

`newConsoleFixtureKeyed(t, withKey)` becomes `return newConsoleFixtureOn(t, withKey, nil)`.

- [ ] **Step 2: Write the test**

Create `internal/console/dbforward_minio_test.go`:

```go
package console

import (
	"bufio"
	"context"
	"encoding/json"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	lss3 "github.com/benbjohnson/litestream/s3"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/dbreads/dbreadstest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// uploadReplica writes a file replica's LTX files into an S3 replica the way
// a box's litestream does (WriteLTXFile), level by level.
func uploadReplica(t *testing.T, from *lsfile.ReplicaClient, to *lss3.ReplicaClient) {
	t.Helper()
	ctx := context.Background()
	for level := 0; level <= litestream.SnapshotLevel; level++ {
		itr, err := from.LTXFiles(ctx, level, 0, false)
		if err != nil {
			t.Fatal(err)
		}
		for itr.Next() {
			f := itr.Item()
			rc, err := from.OpenLTXFile(ctx, level, f.MinTXID, f.MaxTXID, 0, 0)
			if err != nil {
				t.Fatal(err)
			}
			if _, err := to.WriteLTXFile(ctx, level, f.MinTXID, f.MaxTXID, rc); err != nil {
				t.Fatal(err)
			}
			rc.Close()
		}
		itr.Close()
	}
}

// dbfGateStore holds the box's chunk and result creates, once armed, until
// release closes, and records when each landed: T16 needs a box that
// publishes only after the console has ended the request (plan review
// finding 19).
type dbfGateStore struct {
	release.Store
	mu      sync.Mutex
	open    chan struct{}
	reached chan struct{}
	once    *sync.Once
	landed  []time.Time
}

func (g *dbfGateStore) arm() {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.open, g.reached, g.once, g.landed = make(chan struct{}), make(chan struct{}), &sync.Once{}, nil
}

func (g *dbfGateStore) release() {
	g.mu.Lock()
	defer g.mu.Unlock()
	close(g.open)
}

func (g *dbfGateStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	g.mu.Lock()
	open, reached, once := g.open, g.reached, g.once
	g.mu.Unlock()
	if open == nil || !(strings.Contains(key, "/out/") || strings.HasSuffix(key, "/result.json")) {
		return g.Store.Put(ctx, key, body, ifMatch)
	}
	once.Do(func() { close(reached) })
	<-open
	etag, err := g.Store.Put(ctx, key, body, ifMatch)
	if err == nil {
		g.mu.Lock()
		g.landed = append(g.landed, time.Now())
		g.mu.Unlock()
	}
	return etag, err
}

func (g *dbfGateStore) landedAfter(t time.Time) (n int, early bool) {
	g.mu.Lock()
	defer g.mu.Unlock()
	for _, at := range g.landed {
		if !at.After(t) {
			early = true
		}
	}
	return len(g.landed), early
}

// TestDBForwardMinio is T1 and T16 against a real S3 server: a console in ie
// and a box in uk sharing one deployment bucket (versioned), each with its
// own data bucket and its own credentials, on one endpoint (spec §12). The
// console's credentials cannot read uk's bucket (a MinIO policy), so a
// console that read it would fail. Two tenants in uk each get their own
// rows. Neither the console nor the box ever deletes or replaces anything
// under db-reads/, and a box's writes after the request ended change
// nothing. It runs only where .amadan/ci.d/30-minio-contract stands MinIO
// up; the box is plan 1's real code joined in process (dbreadstest).
func TestDBForwardMinio(t *testing.T) {
	if os.Getenv("CARLOS_MINIO_TEST") != "1" {
		t.Skip("set CARLOS_MINIO_TEST=1 (with a running MinIO and the DBF_* variables) to run; see .amadan/ci.d/30-minio-contract")
	}
	need := func(name string) string {
		v := os.Getenv(name)
		if v == "" {
			t.Fatalf("%s is not set", name)
		}
		return v
	}
	endpoint := need("MINIO_ENDPOINT")
	rootKey, rootSecret := need("DBF_ROOT_KEY"), need("DBF_ROOT_SECRET")
	consoleKey, consoleSecret := need("DBF_CONSOLE_KEY"), need("DBF_CONSOLE_SECRET")
	boxKey, boxSecret := need("DBF_BOX_KEY"), need("DBF_BOX_SECRET")
	store := func(key, secret, bucket string) *release.S3Store {
		return &release.S3Store{Bucket: bucket, Client: awss3.New(awss3.Options{
			Region: "us-east-1", Credentials: credentials.NewStaticCredentialsProvider(key, secret, ""),
			BaseEndpoint: aws.String(endpoint), UsePathStyle: true,
			RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
			ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
		})}
	}
	replica := func(key, secret, bucket, prefix string) *lss3.ReplicaClient {
		c := lss3.NewReplicaClient()
		c.AccessKeyID, c.SecretAccessKey, c.Region, c.Bucket, c.Path, c.Endpoint, c.ForcePathStyle =
			key, secret, "us-east-1", bucket, prefix, endpoint, true
		return c
	}
	run := strconv.FormatInt(time.Now().UnixNano(), 36)
	shop, other := "shop-"+run+".example.com", "other-"+run+".example.com"
	for host, batches := range map[string]int{shop: 3, other: 1} {
		dir := t.TempDir()
		dbdoortest.Replicate(t, dir, batches)
		uploadReplica(t, lsfile.NewReplicaClient(dir), replica(rootKey, rootSecret, "dbf-data-uk", "instances/"+host))
	}

	// The policy, shown: the console's credentials cannot list uk's bucket.
	itr, err := replica(consoleKey, consoleSecret, "dbf-data-uk", "instances/"+shop).LTXFiles(context.Background(), 0, 0, false)
	if err == nil {
		for itr.Next() {
		}
		err = itr.Err()
		itr.Close()
	}
	if err == nil {
		t.Fatal("the console's credentials read uk's data bucket: the test proves nothing")
	}

	deploy := store(consoleKey, consoleSecret, "dbf-deploy")
	fx := newConsoleFixtureOn(t, true, strictStore{Store: deploy, t: t})
	fx.srv.cfg.DBForward, fx.srv.cfg.Region, fx.srv.cfg.HibernateBucket = true, "ie", "dbf-data-ie"
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		return replica(consoleKey, consoleSecret, "dbf-data-ie", prefix), nil // the only data bucket this console has
	}
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "member@example.com", "code-minio")
	for _, host := range []string{shop, other} {
		putShopRecord(t, fx, provision.Record{Host: host, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	}
	putEdgesFile(t, fx.rel, "uk-"+run, "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-"+run, time.Now())
	gate := &dbfGateStore{Store: store(boxKey, boxSecret, "dbf-deploy")}
	box := dbreadstest.Start(t, dbreadstest.Options{
		Store: strictStore{Store: gate, t: t}, Region: "uk", Box: "uk-" + run,
		Pub: fx.pub, HibernatePrefix: "instances",
		OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) {
			return replica(boxKey, boxSecret, "dbf-data-uk", prefix), nil
		},
	})
	runBox(t, box)
	dfx := &dbfFixture{consoleFixture: fx, member: member}

	// T1: a forwarded query end to end, for each tenant, each its own rows.
	for host, want := range map[string]string{shop: "n\n30\n", other: "n\n10\n"} {
		body, key := queryBody(t, "SELECT count(*) AS n FROM t")
		res := dfx.openRead(t, host, member, body)
		lines := readLines(t, res.Body)
		res.Body.Close()
		if rows, _ := openAnswer(t, key, lines); string(rows) != want {
			t.Fatalf("%s: %q, want %q", host, rows, want)
		}
	}

	// T16 (plan review finding 19): the real box reaches publication and is
	// held there; the console's request ends, by disconnect and then by its
	// bound; only then does the box publish. Its late chunks and result land
	// after the console ended, change nothing the console sends, and the
	// strict stores see no delete and no replacing write from either side.
	late := func(t *testing.T, end func(res *http.Response, br *bufio.Reader)) {
		gate.arm()
		body, _ := queryBody(t, "SELECT count(*) AS n FROM t")
		res := dfx.openRead(t, shop, member, body)
		br := bufio.NewReader(res.Body)
		line, _ := br.ReadBytes('\n')
		var adm dbreads.Line
		json.Unmarshal(line, &adm)
		end(res, br)
		// The handler's last act is the cancel: once it is there, the
		// console has ended the request.
		waitKey := func(key string) {
			deadline := time.Now().Add(60 * time.Second)
			for {
				if _, _, err := deploy.Get(context.Background(), key); err == nil {
					return
				}
				if time.Now().After(deadline) {
					t.Fatalf("no %s", key)
				}
				time.Sleep(100 * time.Millisecond)
			}
		}
		waitKey(dbreads.CancelKey(adm.Admitted))
		consoleEnded := time.Now()
		select {
		case <-gate.reached:
		case <-time.After(60 * time.Second):
			t.Fatal("the box never reached publication")
		}
		if keys, _ := deploy.List(context.Background(), dbreads.OutPrefix(adm.Admitted)); len(keys) != 0 {
			t.Fatalf("output published before the console ended: %v", keys)
		}
		gate.release()
		waitKey(dbreads.ResultKey(adm.Admitted))
		// At least the result lands late; a held chunk lands too unless the
		// owner's cancel watch ended its create first (plan 1's watchCancel).
		if n, early := gate.landedAfter(consoleEnded); n < 1 || early {
			t.Fatalf("%d late writes, any before the console ended: %v", n, early)
		}
	}
	t.Run("by disconnect", func(t *testing.T) {
		late(t, func(res *http.Response, _ *bufio.Reader) {
			select {
			case <-gate.reached:
			case <-time.After(60 * time.Second):
				t.Fatal("the box never reached publication")
			}
			res.Body.Close()
		})
	})
	t.Run("by timeout", func(t *testing.T) {
		// The request's own bound, well past the box's restore and query of
		// three batches, well short of the box's 20-minute owner deadline.
		fx.srv.dbf.bound, fx.srv.dbf.grace = 20*time.Second, 0
		late(t, func(res *http.Response, br *bufio.Reader) {
			defer res.Body.Close()
			lines := readLines(t, br)
			if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
				t.Fatalf("by timeout: %+v", last)
			}
			for _, l := range lines {
				if l.Chunk != 0 || l.Result != nil {
					t.Fatalf("a held box's output reached the console: %+v", l)
				}
			}
		})
	})
}
```

- [ ] **Step 3: Add the run to the MinIO step**

In `.amadan/ci.d/30-minio-contract`, before the final `echo "minio contract: PASS"`, add:

```sh
# The forwarded db read (db region forward plan 2: T1 and T16): one
# deployment bucket, versioned, and a data bucket per region, on the one
# server. The console user cannot read uk's data bucket; the uk box user
# can. Neither may delete: the protocol deletes nothing (J4), so a delete
# would fail at the server as well as in the tests' store wrappers.
"$mc" --quiet mb local/dbf-deploy local/dbf-data-ie local/dbf-data-uk
"$mc" --quiet version enable local/dbf-deploy
cat >"$rundir/dbf-console.json" <<'EOF'
{"Version": "2012-10-17", "Statement": [
  {"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::dbf-deploy", "arn:aws:s3:::dbf-data-ie"]},
  {"Effect": "Allow", "Action": ["s3:GetObject", "s3:PutObject"], "Resource": ["arn:aws:s3:::dbf-deploy/*"]},
  {"Effect": "Allow", "Action": ["s3:GetObject"], "Resource": ["arn:aws:s3:::dbf-data-ie/*"]}
]}
EOF
cat >"$rundir/dbf-ukbox.json" <<'EOF'
{"Version": "2012-10-17", "Statement": [
  {"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::dbf-deploy", "arn:aws:s3:::dbf-data-uk"]},
  {"Effect": "Allow", "Action": ["s3:GetObject", "s3:PutObject"], "Resource": ["arn:aws:s3:::dbf-deploy/*"]},
  {"Effect": "Allow", "Action": ["s3:GetObject"], "Resource": ["arn:aws:s3:::dbf-data-uk/*"]}
]}
EOF
"$mc" --quiet admin policy create local dbf-console "$rundir/dbf-console.json"
"$mc" --quiet admin policy create local dbf-ukbox "$rundir/dbf-ukbox.json"
"$mc" --quiet admin user add local dbfconsole dbfconsolesecret1
"$mc" --quiet admin user add local dbfukbox dbfukboxsecret1
"$mc" --quiet admin policy attach local dbf-console --user dbfconsole
"$mc" --quiet admin policy attach local dbf-ukbox --user dbfukbox

out3=$rundir/test-dbf.log
set +e
CARLOS_MINIO_TEST=1 \
	MINIO_ENDPOINT="http://127.0.0.1:$port" \
	DBF_ROOT_KEY="$root_user" DBF_ROOT_SECRET="$root_pass" \
	DBF_CONSOLE_KEY=dbfconsole DBF_CONSOLE_SECRET=dbfconsolesecret1 \
	DBF_BOX_KEY=dbfukbox DBF_BOX_SECRET=dbfukboxsecret1 \
	go test ./internal/console/ -run 'TestDBForwardMinio' -count=1 -v >"$out3" 2>&1
test_status=$?
set -e
cat "$out3"
if [ "$test_status" -ne 0 ]; then
	exit "$test_status"
fi
if ! grep -q '^--- PASS: TestDBForwardMinio (' "$out3"; then
	echo >&2
	echo "the forwarded db read test did not PASS: it was skipped, filtered out, or renamed." >&2
	echo "It skips unless CARLOS_MINIO_TEST=1. Check the env block above and that" >&2
	echo "TestDBForwardMinio still exists in internal/console." >&2
	exit 1
fi
```

If the console fixture's setup needs a delete under the deployment bucket (it should not), the test fails with AccessDenied: add `s3:DeleteObject` on `dbf-deploy/*` to the console policy only for the prefixes the fixture deletes, never `db-reads/*`, and say why in the policy's comment.

- [ ] **Step 4: Run it against a local MinIO**

Run the MinIO step locally only if a MinIO binary is already in `.amadan-cache/` from an earlier run; do not download one without asking (machine rule 5). Otherwise push and read the step's output in CI: `amadan ci status carlos/platform -branch db-region-forward-2-query`, then the step log. Expected: `--- PASS: TestDBForwardMinio`.

- [ ] **Step 5: Commit**

```bash
git add internal/console/dbforward_minio_test.go internal/console/server_test.go .amadan/ci.d/30-minio-contract
git commit -m "db forward: T1 and T16 against MinIO with per-region credentials"
```

---

### Task 13: The `db-reads/` lifecycle rule (J4, T17) and its approval plan

**Files:**
- Modify: `infra/modules/deployment-buckets/main.tf:28-50`
- Create: `infra/modules/deployment-buckets/tests/db_reads.tftest.hcl`
- Create: `docs/superpowers/notes/db-reads-lifecycle-approval.md`

- [ ] **Step 1: Write the failing module test**

Create `infra/modules/deployment-buckets/tests/db_reads.tftest.hcl`:

```hcl
# db region forward spec T17 (J4): the deployment bucket expires everything
# under db-reads/: current versions a day after they are written,
# noncurrent versions a day after they become noncurrent, and expired
# delete markers. It is the forwarded read protocol's only deletion.

mock_provider "aws" {}

variables {
  deployment_bucket_name = "dep"
  data_bucket_name       = "data"
}

run "db_reads_expire_with_their_versions_and_markers" {
  command = plan

  assert {
    condition = length([
      for r in aws_s3_bucket_lifecycle_configuration.deployment.rule : r
      if r.id == "db-reads-expire" && r.status == "Enabled" &&
      one(r.filter).prefix == "db-reads/" &&
      one(r.expiration).days == 1 &&
      one(r.noncurrent_version_expiration).noncurrent_days == 1
    ]) == 1
    error_message = "db-reads/ must expire current versions after 1 day and noncurrent versions 1 day after they become noncurrent"
  }

  assert {
    condition = length([
      for r in aws_s3_bucket_lifecycle_configuration.deployment.rule : r
      if r.id == "db-reads-delete-markers" && r.status == "Enabled" &&
      one(r.filter).prefix == "db-reads/" &&
      one(r.expiration).expired_object_delete_marker == true
    ]) == 1
    error_message = "db-reads/ must remove expired delete markers"
  }

  assert {
    condition = length([
      for r in aws_s3_bucket_lifecycle_configuration.deployment.rule : r
      if r.id == "geoip-noncurrent" && one(r.filter).prefix == "control/geoip/"
    ]) == 1
    error_message = "the control/geoip/ rule must stay as it was"
  }
}
```

Run: `tofu -chdir=infra/modules/deployment-buckets init -backend=false && tofu -chdir=infra/modules/deployment-buckets test -filter=tests/db_reads.tftest.hcl`
Expected: FAIL on the first two asserts. (`init` downloads the AWS provider from the OpenTofu registry; in the sandbox declare `registry.opentofu.org` and the provider's download host as allowed domains, or run it with the sandbox off.)

- [ ] **Step 2: Add the rules**

In `infra/modules/deployment-buckets/main.tf`, change the comment above `aws_s3_bucket_lifecycle_configuration.deployment` so it no longer says nothing else expires, and add two rules after `geoip-noncurrent`:

```hcl
  # Forwarded db reads (db region forward spec §5.7, J4): every object under
  # db-reads/ (requests, queue entries, picked markers, sealed chunks,
  # results, cancels) expires a day after it is written, each noncurrent
  # version a day after it becomes noncurrent, and expired delete markers
  # go. The protocol deletes nothing, so these rules are its only deletion;
  # a read's ciphertext lives until they act, typically two to four days.
  rule {
    id     = "db-reads-expire"
    status = "Enabled"
    filter {
      prefix = "db-reads/"
    }
    expiration {
      days = 1
    }
    noncurrent_version_expiration {
      noncurrent_days = 1
    }
    abort_incomplete_multipart_upload {
      days_after_initiation = 1
    }
  }

  # S3 refuses an expired-object-delete-marker expiration beside days in one
  # rule, so the markers have a rule of their own.
  rule {
    id     = "db-reads-delete-markers"
    status = "Enabled"
    filter {
      prefix = "db-reads/"
    }
    expiration {
      expired_object_delete_marker = true
    }
  }
```

- [ ] **Step 3: Run the module tests**

Run: `tofu -chdir=infra/modules/deployment-buckets test`
Expected: PASS for `db_reads.tftest.hcl` and the module's two existing test files.

- [ ] **Step 4: Write the approval plan**

Create `docs/superpowers/notes/db-reads-lifecycle-approval.md`:

```markdown
# Approval plan: the db-reads/ lifecycle rule on the deployment buckets

Status: PREPARED, not applied. AGENTS.md's production rule applies: nothing
here is applied until Paul approves this scope.

## What and why

Two lifecycle rules on each deployment bucket, both filtered to the prefix
`db-reads/` (db region forward spec §5.7, §11.1, J4):

- `db-reads-expire`: current versions expire 1 day after creation;
  noncurrent versions expire 1 day after they become noncurrent; incomplete
  multipart uploads are aborted after 1 day.
- `db-reads-delete-markers`: expired object delete markers are removed.

The forwarded read protocol writes create-only objects under `db-reads/`
and deletes nothing. These rules are its only deletion. They must be in
place on a deployment's bucket before any console there sets
`CARLOS_DB_FORWARD=1`; without them `db-reads/` objects are never removed.

## Scope

| | Dev | Flagship |
|---|---|---|
| Account and region | the dev deployment's account, eu-west-1 | the flagship deployment's account, eu-west-1 |
| Root | `platform-infrastructure` (tracks `main`) | `carlosframework/carloku-infrastructure` (modules at a release tag) |
| Resource | `aws_s3_bucket_lifecycle_configuration.deployment` inside the `deployment-buckets` module | the same |
| Change | two rules added; the `control/geoip/` rule unchanged | the same |

No IAM change, no new permission, no traffic change. Nothing outside
`db-reads/` is affected: the filter is the prefix, and no other key in the
bucket starts with it.

## Cost

Lifecycle expiration is free. Storage of `db-reads/` objects is at most a
few days of forwarded reads (a query's ciphertext is at most 64 MiB).

## Steps

Dev first (Paul's roll order), then the flagship:

1. Dev: in `platform-infrastructure`, `tofu init -upgrade`, then
   `tofu plan -target=<the deployment-buckets module address>` (find it
   with `grep -n 'deployment-buckets' *.tf`). The plan must show only the
   lifecycle configuration changing, with the two rules added. Apply with
   the same `-target` after approval.
2. Flagship: cut the platform release that carries this module change,
   push its tag to the GitHub mirror (`git push
   https://github.com/carlosframework/platform.git refs/tags/vX.Y.Z`), bump
   every `?ref=` in `carloku-infrastructure`, `tofu init -upgrade`, then the
   same targeted plan. Apply with the same `-target` after approval. A
   whole-root apply is not allowed (user_data drift restarts every box).

## Verification

`aws s3api get-bucket-lifecycle-configuration --bucket <deployment bucket>
--profile <the deployment's profile>` lists `geoip-noncurrent`,
`db-reads-expire` and `db-reads-delete-markers`, the last two filtered to
`db-reads/`.

## Rollback

Remove the two rules and apply the same targeted change. Until a rule is
back, `db-reads/` objects are not removed, so turn forwarding off
(`CARLOS_DB_FORWARD` unset on the console) first.

## Duration

Permanent while forwarding is in use. Plan 5 (spec §14) moves this traffic
to a bucket per region, which carries its own copy of these rules; these
then stay for any object written before the move.
```

- [ ] **Step 5: Commit**

```bash
git add infra/modules/deployment-buckets docs/superpowers/notes/db-reads-lifecycle-approval.md
git commit -m "deployment-buckets: db-reads/ expires (J4, T17); the approval plan for applying it"
```

---

### Task 14: Mutation checks, census, test weights, merge, CI, and the roll

**Files:**
- Create: `internal/provision/testdata/plan14-root-reads-db-region-forward-2.tsv`
- Modify: `internal/provision/plan14_root_reads_test.go` (load it last, with the seal)
- Create: `internal/provision/db_region_forward_2_census_test.go`
- Modify: `internal/provision/plan9_b8_census_test.go:369`
- Create: `docs/superpowers/notes/census-proposal-db-region-forward-2.md`
- Modify: `.amadan/test-weights.tsv`

- [ ] **Step 1: Mutation checks (spec §11.4)**

In a copy of the tree (`cp -a . "$TMPDIR/dbf2-mut"`), make each mutation, run its test there, record red, revert (`git -C "$TMPDIR/dbf2-mut" checkout -- <file>`). Remove the copy by name afterwards.

| Mutation | Test that must fail |
|---|---|
| in `dbreads` `attempt`, open the replica at a prefix from anything but `rec.Host` (for example the first host the agent ever checked) | `TestDBForwardMinio` (the second tenant's rows), in the MinIO step; `TestDBReadsParityWithTheInRegionDoor` |
| `openDBForward` skips the byte count and SHA-256 comparison | `TestForwardedQueryRefusesAnAnswerThatDoesNotCheckOut`, `TestDBForwardPrintsNothingForABadAnswer` |
| `admitDBRead` releases the counts before returning, or `handleAPIDBRead` drops its `defer` | `TestDBReadAdmissionBounds` |
| `handleAPIDBRead` drops the post-admission `SetReadDeadline` (plan review finding 38) | `TestDBReadSetsBothReadDeadlines` (no deadline past 22 minutes); a finished body lets net/http clear the read deadline itself, so `TestDBForwardLongRequestThroughTheEdge` is not its witness |
| `dbLineWriter.write` drops `Flush` | `TestDBReadLinesAreFlushedAsWritten`, `TestDBForwardLongRequestThroughTheEdge`, `TestDBForwardThroughTheBackhaulHop` |
| `streamDBRead` writes the result line without waiting for `next > res.LastSeq` (plan review finding 31) | `TestDBReadResultWaitsForAChunkTheListingMissed` |
| `streamDBRead` lists chunks before reading `result.json`, the wait left in place (plan review finding 20) | `TestDBReadResultWaitsForItsLastChunk` (the operation-order check: a `list` not preceded by its round's `result`) |
| `admitDBRead` writes the audit row after the creates (plan review finding 11) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("the audit row is down before the request") |
| `admitDBRead` drops any one of its four `ended()` checks (plan review finding 28) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel`, the "ended before <step>" case for that step |
| `admitDBRead` drops `cancelDBRead` on a published step | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("ended before queue", "ended before serial": no cancel) |
| `admitDBRead` calls `s.bumpSerial` instead of `announceDBRead` (plan review finding 33) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("ended during the serial read": the detached write lands) |
| `admitDBRead` drops the U+FFFD refusal (plan review finding 36) | `TestDBReadRouteRefusesBeforeAdmitting` (the two raw bodies) |
| `dbQueryForward` drops the UTF-8 check | `TestForwardedQueryRefusesSQLThatIsNotUTF8`, `TestDBForwardRefusesSQLThatIsNotUTF8` |
| `dbfChunkBound.admit` drops its size floor, its empty-chunk rule or its count bound (plan review finding 37) | `TestForwardedQueryBoundsChunkCountAndSize`, the matching case |
| `streamDBRead` drops the `ownCtx` check before the result line (plan review finding 35) | `TestDBReadResultAfterTheDeadlineIsNotRelayed` |
| `handleAPIDBRead` drops the `preHeader` read deadline (plan review finding 13) | `TestDBReadStalledBodyIsCutAtThePreHeaderBound` |
| `streamDBRead` calls the store on `r.Context()` without `dbfCall`, or drops either phase's deadline context (plan review finding 12) | `TestDBReadStalledStoreKeepsTheStreamAndItsBound` |
| `streamDBRead` relays without the owner's deadline (chunk Gets on `r.Context()`, no `ownCtx` check after them) | `TestDBReadSlowRelayEndsAtTheBound` |
| a startup pass that writes under `db-reads/` | `TestDBReadAdmissionBounds` (the restart) |
| `cancelDBRead` deletes the read's objects, or any console `Put` under `db-reads/` passes an etag | `TestDBReadLateWritesChangeNothing`, `TestDBForwardMinio` |
| `dbForwardEvidence` checks `Complete` and not `BoxesComplete` | `TestDBForwardEvidenceRefusesOnlyOnCompleteEvidence` |

Record the table with every row red in the branch description.

- [ ] **Step 2: Merge main and run the census**

```bash
git fetch origin && git merge origin/main && go build ./...
```

Then add the temporary dump test from plan 1 Task 17 Step 2 (prefixes: `db.go`, `dbforward.go`, `cmd/console/`, `internal/console/apidb`, `internal/console/httpserver.go`, `internal/console/authcli.go`, `internal/console/server.go`, `internal/dbdoor/`, `internal/dbreads/`, `infra/modules/deployment-buckets/`), run it with `TestPlan14RootReadCensus$` and `TestPlan9B8RecordWriteCensus$`, and keep the log under `$TMPDIR`.

- [ ] **Step 3: Write the overlay**

`internal/provision/testdata/plan14-root-reads-db-region-forward-2.tsv`, in plan 1 Task 17 Step 3's formats (each `@legacy` fingerprint computed by plan 1's `TestZZDBForwardFingerprints`, in the temporary dump test, from the row's own fields), with `rev3=PROPOSED-db-region-forward-2` and these bases by identity:

- console units (`Server.handleAPIDBRead`, `Server.admitDBRead`, `Server.streamDBRead`, `dbLineWriter.write`, `dbLineWriter.tick`, `dbfCall`, `Server.readDBResult`, `Server.cancelDBRead`, `Server.announceDBRead`, `Server.dbResidency`, `dbForwardEvidence`, `NewHTTPServer`, and any changed console legacy unit the census names): `accepted-legacy;rev3=PROPOSED-db-region-forward-2;basis=route-and-input (J1,J4, as main's db door units); the console (cmd/console serve under console.service, user carlos-console): <one clause on what the unit reads: the instance record and the edges and status listing for the gate, or the read's picked, result.json and chunk objects in the deployment bucket, relayed to the member as ciphertext>; never reads another region's data bucket, never runs SQL, makes no root decision`;
- CLI units (`dbQueryForward`, `readDBForward`, `openDBForward`, and the changed `dbRestore`, `dbQuery`, `dbPull`): `accepted-legacy;rev3=PROPOSED-db-region-forward-2;basis=root-input (J4, as main's db.go units); the carlos CLI on a person's machine, as the person (no box unit): <one clause>`;
- `internal/dbdoor` `doorError` (changed): re-frozen at main's disposition with "and the db region forward marker's three additive fields" appended;
- `cmd/console` `serve` (changed): re-frozen at its disposition with "CARLOS_DB_FORWARD and console.NewHTTPServer" appended;
- `infra/modules/deployment-buckets/main.tf` (changed, external): re-frozen at `accepted-legacy;rev3=J4;basis=root-input` with "; the db-reads/ lifecycle rules (db region forward, J4), PROPOSED-db-region-forward-2" appended;
- envelopes re-pinned for every directory the census names.

Load it last in `TestPlan14RootReadCensus` with the previous seal checked first, as plan 1 Task 17 Step 4 does, and set the new seal.

- [ ] **Step 4: The 9B8 entries and the claim routes**

Create `internal/provision/db_region_forward_2_census_test.go`:

```go
package provision

// Record-write census entries for db region forward plan 2, layered on
// dbRegionForward1WriteCensusInventory (or whatever TestPlan9B8RecordWriteCensus
// calls on the merged main). PROPOSED-db-region-forward-2, asked of the census
// owner on #280. The route is one more key-site line in registerCLIRoutes; the
// request, the queue entry and cancel are conditional creates under db-reads/
// in the deployment bucket, never an instance or sidecar record: nonrecord.
func dbRegionForward2WriteCensusInventory() []censusEntry {
	entries := append([]censusEntry(nil), dbRegionForward1WriteCensusInventory()...)
	replacements := []censusEntry{
		{"internal/console/authcli.go", "Server.registerCLIRoutes", "key-site", "<kinds>", "<digest>"},
		{"internal/console/apidbforward.go", "Server.admitDBRead", "nonrecord", "<kinds>", "<digest>"},
		{"internal/console/apidbforward.go", "Server.cancelDBRead", "nonrecord", "<kinds>", "<digest>"},
	}
	for _, r := range replacements {
		found := false
		for i, e := range entries {
			if e.File == r.File && e.Unit == r.Unit {
				entries[i], found = r, true
				break
			}
		}
		if !found {
			entries = append(entries, r)
		}
	}
	return entries
}
```

with each `<kinds>` and `<digest>` from the dump's `WRITE` lines, and point `TestPlan9B8RecordWriteCensus` at it. Check `route POST /api/cli/apps/{account}/{app}/db/{host}/reads` is in `internal/console/testdata/claim-routes.txt` (Task 6).

Run: `go test ./internal/provision/ -run 'TestPlan14RootReadCensus$|TestPlan9B8RecordWriteCensus$|TestPlan14' -count=1` and `go test ./internal/console/ -run 'Claim.*Route' -count=1`. Expected: PASS. Delete the dump test.

- [ ] **Step 5: The census proposal**

Write `docs/superpowers/notes/census-proposal-db-region-forward-2.md` (plan 1's form) and post it on #280.

- [ ] **Step 6: Test weights**

Measure the new top-level tests without a full run:

```bash
go test -count=1 -json -run 'TestDBForward|TestDBRead|TestListDoor' ./internal/console/ > "$TMPDIR/dbf2-console.json"
go test -count=1 -json -run 'TestDBForward|TestForwardedQuery' . > "$TMPDIR/dbf2-root.json"
grep '"Action":"pass"' "$TMPDIR/dbf2-console.json" "$TMPDIR/dbf2-root.json" | grep -v '"Test":"[^"]*/' | grep -o '"Test":"[^"]*","Elapsed":[0-9.]*'
rm "$TMPDIR/dbf2-console.json" "$TMPDIR/dbf2-root.json"
```

Add a row (`<package>	<test>	<seconds>`) to `.amadan/test-weights.tsv` for every test of a second or more, in descending order within its package, keeping the header. `TestDBForwardPrintsNothingForABadAnswer` (about a minute) and `TestDBForwardLongRequestThroughTheEdge` (about 45 s) will be the heaviest root rows.

- [ ] **Step 7: Final checks, push, describe, CI**

```bash
gofmt -l . ; go vet . ./internal/console/ ./internal/dbreads/ ./internal/dbdoor/ ./cmd/console/
go build ./... && GOOS=darwin GOARCH=arm64 go build ./... && GOOS=windows GOARCH=amd64 go build ./...
git add internal/provision docs/superpowers/notes/census-proposal-db-region-forward-2.md .amadan/test-weights.tsv
git commit -m "census: db-region-forward-2 overlay (PROPOSED), 9B8 entries; test weights"
git push origin db-region-forward-2-query
amadan branch describe carlos/platform db-region-forward-2-query -body - < "$TMPDIR/dbf-2-description.md"
amadan ci status carlos/platform -branch db-region-forward-2-query
```

The description names: what plan 2 adds and that a console with `CARLOS_DB_FORWARD` unset is byte-identical to main; the spec and plan; the mutation table; the MinIO step's PASS; whether `TestDBForwardTwentyOneMinuteRead` has been run (once, before handoff, with `CARLOS_DBF_LONG=1`, and its result); the census proposal and the ruling owed on #280; the copy state; the lifecycle approval plan and that nothing is applied. Expected: CI green on the tip.

- [ ] **Step 8: Review, then merge**

Whole-branch review (Astra by a committed brief, or Opus). After fixes and the census ruling: merge main again, rebuild the census if main moved, push, wait for green, squash, and merge only with `git merge-base --is-ancestor origin/main HEAD && amadan branch merge carlos/platform db-region-forward-2-query -expect <tip> -message "..."`. Read main's CI afterwards.

- [ ] **Step 9: The roll (spec §11.3), each live step on Paul's go**

1. Ask Paul to approve `docs/superpowers/notes/db-reads-lifecycle-approval.md` for dev; apply it on dev's deployment bucket and verify.
2. Dev boxes: they carry plan 1 and publish `db-read-1` already; confirm in their edges files.
3. Dev console: release and roll it with plan 2 (Paul's recipe: dev boxes, then dev console), then set `CARLOS_DB_FORWARD=1` in its environment (dev's `console.service` lives only on the box; see the dev console unit note) and restart it.
4. Check: from a laptop signed in to the dev console, `carlos db query --app <an app with an instance on the dev uk box> --instance <that instance> --sql "SELECT count(*) FROM sqlite_master"` and the same with `--at` a few minutes ago. Both print rows; stderr shows DBF-11 and the status line. Then a read of an in-region instance, unchanged.
5. Report on the branch's discussion: plan 1's LIST rate (Task 17 Step 11) and, read-only, the flagship's real database sizes per region (the restored size of each instance's newest snapshot: the last planned file's header, or `aws s3 ls --summarize` of each instance's `ltx/9/` prefix with the flagship's profile) against the 2 GiB cap (ruling 2).
6. Flagship, only on Paul's explicit go after the dev check: the lifecycle rule (its own approval), the boxes, the console, then `CARLOS_DB_FORWARD=1` on the flagship console. Verify with one forwarded query of a moved instance in a test account, never a tenant's data.