db: refuse a pending migration older than the newest applied one

applyMigrations now fails closed, before any write, when a pending migration has a version below the highest version recorded in schema_migrations: migrations must be appended, never inserted.
db: versioned migrations runner with schema_migrations
2026-11-14 20:20:54 +08:00 · 2026-11-14 13:10:57 +01:00 · 2026-11-14 12:52:34 +01:00 · 2026-11-14 12:27:39 +01:00 · 2026-11-14 12:17:33 +01:00 · 2026-11-14 12:06:44 +01:00
29 changed files with 1597 additions and 195 deletions
@@ -114,6 +114,7 @@ func bbotTyped(raw bbotRawEvent) bbotEvent {
 	}
 	if u, err := url.Parse(strings.TrimSpace(raw.URL)); err == nil && ev.Host != "" &&
 		(u.Scheme == "http" || u.Scheme == "https") && strings.EqualFold(u.Hostname(), ev.Host) {
+		u.User = nil // credentials never reach an asset
 		ev.URL = u.String()
 	}
 	return ev
@@ -71,3 +71,15 @@ func TestParseBbotJSONLTypedOnly(t *testing.T) {
 		}
 	}
 }
+
+// The typed url of an in-scope URL event loses its userinfo, keeps its query.
+func TestParseBbotURLUserinfo(t *testing.T) {
+	res, err := parseBbotJSONL(strings.NewReader(`{"type":"URL","host":"www.example.com","url":"https://scanuser:s3cr3t@www.example.com/app?id=7","scope_distance":0}` + "\n"))
+	if err != nil || len(res.InScope) != 1 {
+		t.Fatalf("parse: %v, %+v", err, res)
+	}
+	want := []assetInputItem{{Type: "service", URL: "https://www.example.com/app?id=7"}}
+	if got := res.assetItems(); !reflect.DeepEqual(got, want) {
+		t.Errorf("assetItems = %+v, want %+v", got, want)
+	}
+}
@@ -83,8 +83,8 @@ func scanHost(h string) string {
 	return ""
 }

-// scanURL returns raw re-serialized and its host when it is an http(s) URL
-// with a valid host, "" otherwise.
+// scanURL returns raw re-serialized without its userinfo, and its host, when
+// it is an http(s) URL with a valid host, "" otherwise.
 func scanURL(raw string) (string, string) {
 	u, err := url.Parse(strings.TrimSpace(raw))
 	if err != nil || (u.Scheme != "http" && u.Scheme != "https") {
@@ -94,9 +94,31 @@ func scanURL(raw string) (string, string) {
 	if h == "" {
 		return "", ""
 	}
+	u.User = nil // credentials never reach an asset, a note or a fact
 	return u.String(), h
 }

+// stripUserinfo removes the userinfo (user:pass@) from raw, a URL or a bare
+// host[:port] as a scanner prints it, and keeps every other byte. It is for
+// the raw values kept in a fact detail: unlike scanURL it never rejects, and a
+// value with no userinfo, or a path with no authority, is returned unchanged.
+func stripUserinfo(raw string) string {
+	start := 0 // the authority follows "scheme://" or "//"; a bare value is all authority
+	if i := strings.Index(raw, "//"); i == 0 || i > 0 && raw[i-1] == ':' && !strings.ContainsAny(raw[:i-1], "/?#@") {
+		start = i + 2
+	} else if strings.ContainsAny(raw, "/?#") {
+		return raw
+	}
+	auth := raw[start:]
+	if j := strings.IndexAny(auth, "/?#"); j >= 0 {
+		auth = auth[:j]
+	}
+	if at := strings.LastIndex(auth, "@"); at >= 0 {
+		return raw[:start] + raw[start+at+1:]
+	}
+	return raw
+}
+
 // scanHostOf reads a host from a URL, a host:port or a bare host.
 func scanHostOf(s string) string {
 	if _, h := scanURL(s); h != "" {
@@ -342,7 +364,7 @@ func (s *scanIngest) addNuclei(line []byte) {
 	}
 	s.note("results", tid+" ("+sev+") on "+host)
 	s.fact(fmt.Sprintf("nuclei: template %s (severity %s) matched on %s; unverified scanner result", tid, sev, host),
-		map[string]any{"name": r.Info.Name, "host": r.Host, "matched_at": r.MatchedAt,
+		map[string]any{"name": r.Info.Name, "host": stripUserinfo(r.Host), "matched_at": stripUserinfo(r.MatchedAt),
 			"matcher_name": r.MatcherName, "extracted_results": r.Extracted})
 }

@@ -394,7 +416,7 @@ func (s *scanIngest) addHttpx(line []byte) {
 	s.note("urls", fmt.Sprintf("%s %d", u, r.StatusCode))
 	s.fact(fmt.Sprintf("httpx: %s responded with status %d", u, r.StatusCode),
 		map[string]any{"title": r.Title, "tech": r.Tech, "webserver": r.Webserver, "content_length": r.ContentLength,
-			"host": r.Host, "ip": r.IP, "tls": r.TLS})
+			"host": stripUserinfo(r.Host), "ip": r.IP, "tls": r.TLS})
 }

 // ---- ffuf -of json ----
@@ -480,5 +502,5 @@ func (s *scanIngest) addFfuf(raw json.RawMessage) {
 	s.Assets = append(s.Assets, item)
 	s.note("urls", fmt.Sprintf("%s %d", u, r.Status))
 	s.fact(fmt.Sprintf("ffuf: %s returned status %d, %d bytes, %d words, %d lines", u, r.Status, r.Length, r.Words, r.Lines),
-		map[string]any{"input": r.Input, "redirectlocation": r.Redirect})
+		map[string]any{"input": r.Input, "redirectlocation": stripUserinfo(r.Redirect)})
 }
@@ -248,6 +248,137 @@ func TestScanIngestJSONLLineLimits(t *testing.T) {
 	}
 }

+// scanURL drops the userinfo and keeps scheme, host, port, path and query; an
+// @ in the path is not userinfo; a URL it refused before is still refused.
+func TestScanURLUserinfo(t *testing.T) {
+	cases := []struct{ raw, url, host string }{
+		{"http://scanuser:s3cr3t@scan-ingest.example.com:8080/app?id=7&x=y", "http://scan-ingest.example.com:8080/app?id=7&x=y", "scan-ingest.example.com"},
+		{"https://scanuser@scan-ingest.example.com/app", "https://scan-ingest.example.com/app", "scan-ingest.example.com"},
+		{"https://scan-ingest.example.com/a@b/c?x=@y", "https://scan-ingest.example.com/a@b/c?x=@y", "scan-ingest.example.com"},
+		{"http://scanuser:s3cr3t@scan-ingest.example.com/%zz", "", ""},
+		{"javascript:alert(1)", "", ""},
+		{"http://scanuser:s3cr3t@bad host/", "", ""},
+		{"not a url", "", ""},
+	}
+	for _, c := range cases {
+		if u, h := scanURL(c.raw); u != c.url || h != c.host {
+			t.Errorf("scanURL(%q) = %q, %q; want %q, %q", c.raw, u, h, c.url, c.host)
+		}
+	}
+}
+
+func TestStripUserinfo(t *testing.T) {
+	cases := map[string]string{
+		"https://scanuser:s3cr3t@scan-ingest.example.com:443/login?next=1": "https://scan-ingest.example.com:443/login?next=1",
+		"http://scanuser:s3cr3t@scan-ingest.example.com/%zz":               "http://scan-ingest.example.com/%zz",
+		"//scanuser:s3cr3t@scan-ingest.example.com/":                       "//scan-ingest.example.com/",
+		"scanuser:s3cr3t@scan-ingest.example.com:443":                      "scan-ingest.example.com:443",
+		"https://scan-ingest.example.com/a@b?x=@y":                         "https://scan-ingest.example.com/a@b?x=@y",
+		"https://scan-ingest.example.com/?r=http://scanuser:s3cr3t@x/":     "https://scan-ingest.example.com/?r=http://scanuser:s3cr3t@x/",
+		"/admin@x/":                   "/admin@x/",
+		"scan-ingest.example.com:443": "scan-ingest.example.com:443",
+		"198.51.100.77":               "198.51.100.77",
+		"":                            "",
+	}
+	for raw, want := range cases {
+		if got := stripUserinfo(raw); got != want {
+			t.Errorf("stripUserinfo(%q) = %q, want %q", raw, got, want)
+		}
+	}
+}
+
+// Scanner lines carrying user:pass: no asset, url, notable item or fact keeps
+// the credentials, and the query survives.
+const (
+	nucleiUserinfoFixture = `{"template-id":"tech-detect","info":{"severity":"info"},"host":"https://scanuser:s3cr3t@scan-ingest.example.com","matched-at":"https://scanuser:s3cr3t@scan-ingest.example.com/login?next=1"}
+{"template-id":"tech-detect","info":{"severity":"info"},"host":"scanuser:s3cr3t@scan-ingest.example.com:443","matched-at":"https://scanuser:s3cr3t@scan-ingest.example.com:443/x"}
+`
+	httpxUserinfoFixture = `{"url":"https://scanuser:s3cr3t@scan-ingest.example.com/app?id=7","status_code":200,"host":"198.51.100.77"}
+`
+	ffufUserinfoFixture = `{"results":[{"input":{"FUZZ":"app"},"status":301,"url":"https://scanuser:s3cr3t@scan-ingest.example.com/app?id=7","redirectlocation":"https://scanuser:s3cr3t@scan-ingest.example.com/app/"}]}
+`
+)
+
+// hasScanCreds reports whether v holds a user or password of the fixtures.
+func hasScanCreds(v string) bool {
+	for _, c := range []string{"s3cr3t", "scanuser", "alice", "bob"} {
+		if strings.Contains(v, c) {
+			return true
+		}
+	}
+	return false
+}
+
+func checkNoUserinfo(t *testing.T, format string, s *scanIngest) {
+	t.Helper()
+	var all []string
+	for _, a := range s.Assets {
+		all = append(all, a.URL)
+	}
+	for u := range s.urls {
+		all = append(all, u)
+	}
+	for _, items := range s.Notable {
+		all = append(all, items...)
+	}
+	for _, f := range s.Facts {
+		d, _ := strconv.Unquote(f.Detail)
+		all = append(all, f.Summary, d)
+	}
+	for _, v := range all {
+		if hasScanCreds(v) {
+			t.Fatalf("%s: credentials kept in %q", format, v)
+		}
+	}
+}
+
+func TestScanIngestStripsUserinfo(t *testing.T) {
+	cases := []struct {
+		format, content string
+		want            string // events urls facts assets
+		url             string // first asset URL
+	}{
+		{"nuclei-jsonl", nucleiUserinfoFixture, "[2 2 2 2]", "https://scan-ingest.example.com"},
+		{"httpx-jsonl", httpxUserinfoFixture, "[1 1 1 1]", "https://scan-ingest.example.com/app?id=7"},
+		{"ffuf-json", ffufUserinfoFixture, "[1 1 1 1]", "https://scan-ingest.example.com/app?id=7"},
+	}
+	for _, c := range cases {
+		s := parseScan(t, c.format, c.content)
+		if got := fmt.Sprint([]int{s.Events, len(s.urls), len(s.Facts), len(s.Assets)}); got != c.want || s.Skipped != 0 {
+			t.Fatalf("%s: events,urls,facts,assets = %s, want %s (skipped %d)", c.format, got, c.want, s.Skipped)
+		}
+		if s.Assets[0].URL != c.url {
+			t.Fatalf("%s: asset URL %q, want %q", c.format, s.Assets[0].URL, c.url)
+		}
+		checkNoUserinfo(t, c.format, s)
+	}
+	// The raw values kept in the fact detail lose only their userinfo.
+	s := parseScan(t, "nuclei-jsonl", nucleiUserinfoFixture)
+	if d, _ := strconv.Unquote(s.Facts[1].Detail); !strings.Contains(d, `"host":"scan-ingest.example.com:443"`) ||
+		!strings.Contains(d, `"matched_at":"https://scan-ingest.example.com:443/x"`) {
+		t.Fatalf("nuclei detail %s", d)
+	}
+	s = parseScan(t, "ffuf-json", ffufUserinfoFixture)
+	if d, _ := strconv.Unquote(s.Facts[0].Detail); !strings.Contains(d, `"redirectlocation":"https://scan-ingest.example.com/app/"`) {
+		t.Fatalf("ffuf detail %s", d)
+	}
+}
+
+// The same URL with different credentials is one URL after normalisation:
+// both events are counted, both assets carry the same URL.
+func TestScanIngestUserinfoDedup(t *testing.T) {
+	s := parseScan(t, "httpx-jsonl", httpxUserinfoDedupFixture)
+	if s.Events != 2 || len(s.urls) != 1 || len(s.Assets) != 2 || s.Assets[0].URL != s.Assets[1].URL ||
+		s.Assets[0].URL != "https://scan-ingest.example.com/app?id=7" {
+		t.Fatalf("events=%d urls=%d assets=%+v", s.Events, len(s.urls), s.Assets)
+	}
+	checkNoUserinfo(t, "httpx-jsonl", s)
+}
+
+const httpxUserinfoDedupFixture = `{"url":"https://alice:s3cr3t-one@scan-ingest.example.com/app?id=7","status_code":200}
+{"url":"https://bob:s3cr3t-two@scan-ingest.example.com/app?id=7","status_code":200}
+`
+
 // Past scanMaxItems assets and facts, records are counted, not ingested.
 func TestScanIngestItemCap(t *testing.T) {
 	var b strings.Builder
@@ -199,3 +199,78 @@ func TestIngestScanOutputJSONPG(t *testing.T) {
 		t.Fatalf("re-ingestion: %d assets and %d facts, first run %d and %d", a, f, assets, facts)
 	}
 }
+
+// ingest_scan_output (PG), credentials in URLs: no asset row and no fact keeps
+// them, and the same URL with different credentials is one asset row and one
+// fact.
+func TestIngestScanOutputUserinfoPG(t *testing.T) {
+	d := testDB(t)
+	t.Cleanup(func() { _ = d.Close() })
+	task, err := d.CreateTaskWithOptions("ingest_scan_output_userinfo", "scan-ingest.example.com", db.TaskCreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	const assetsWhere = `domain LIKE '%scan-ingest.example.com' OR url LIKE '%scan-ingest.example.com%'`
+	t.Cleanup(func() {
+		_ = d.DeleteTask(task.ID)
+		_, _ = d.Exec(`DELETE FROM assets WHERE ` + assetsWhere)
+	})
+	store := d.Exploration(task.ExplorationID)
+	tsx := NewToolSet(store, "worker-1")
+	tsx.SetTaskID(task.ID)
+	tsx.SetAssetStore(d.Assets(), d.Companies())
+	dir := t.TempDir()
+	tool := tsx.ingestScanOutput()
+	ingest := func(format, content string) map[string]any {
+		t.Helper()
+		if err := os.WriteFile(filepath.Join(dir, "scan.out"), []byte(content), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		in, _ := json.Marshal(map[string]string{"format": format, "path": "scan.out"})
+		res, err := tool.Call(context.Background(), in, &actool.ToolContext{WorkingDir: dir})
+		var out map[string]any
+		if err != nil || res.IsError || json.Unmarshal([]byte(res.Flatten()), &out) != nil ||
+			out["asset_errors"] != nil || out["fact_errors"] != nil {
+			t.Fatalf("%s: %v %s", format, err, res.Flatten())
+		}
+		return out
+	}
+	count := func(q string) int {
+		t.Helper()
+		var n int
+		if err := d.QueryRow(q).Scan(&n); err != nil {
+			t.Fatal(err)
+		}
+		return n
+	}
+
+	out := ingest("httpx-jsonl", httpxUserinfoDedupFixture)
+	if out["events"] != 2.0 || out["urls"] != 1.0 || out["facts_written"] != 1.0 || out["facts_existing"] != 1.0 {
+		t.Fatalf("same URL, different credentials: %v", out)
+	}
+	const appURL = `SELECT count(*) FROM assets WHERE type='service' AND url='https://scan-ingest.example.com/app?id=7'`
+	if n := count(appURL); n != 1 {
+		t.Fatalf("%d asset rows for the normalised URL, want 1", n)
+	}
+	ingest("httpx-jsonl", httpxUserinfoFixture)
+	ingest("ffuf-json", ffufUserinfoFixture)
+	ingest("nuclei-jsonl", nucleiUserinfoFixture)
+	if n := count(appURL); n != 1 {
+		t.Fatalf("%d asset rows for the normalised URL after httpx and ffuf, want 1", n)
+	}
+	if n := count(`SELECT count(*) FROM assets a WHERE (` + assetsWhere + `) AND row_to_json(a)::text ~ 's3cr3t|scanuser|alice|bob'`); n != 0 {
+		t.Fatalf("%d asset rows keep credentials", n)
+	}
+	facts, err := store.ListByKind(db.KindFact, 500)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(facts) == 0 {
+		t.Fatal("no facts written")
+	}
+	for _, f := range facts {
+		if hasScanCreds(string(f.Payload)) {
+			t.Fatalf("fact keeps credentials: %s", f.Payload)
+		}
+	}
+}
@@ -160,7 +160,8 @@ func calcCSegment(ipStr string) string {
 	return ipnet.String()
 }

-// normalizeURL lowercases scheme and host, strips trailing slash from bare roots.
+// normalizeURL lowercases scheme and host, drops userinfo (credentials must
+// never reach assets.url), and strips trailing slash from bare roots.
 func normalizeURL(raw string) string {
 	u, err := url.Parse(raw)
 	if err != nil {
@@ -168,6 +169,7 @@ func normalizeURL(raw string) string {
 	}
 	u.Scheme = strings.ToLower(u.Scheme)
 	u.Host = strings.ToLower(u.Host)
+	u.User = nil
 	s := u.String()
 	if strings.HasSuffix(s, "/") && u.Path == "/" && u.RawQuery == "" {
 		s = strings.TrimSuffix(s, "/")
@@ -893,6 +893,10 @@ func TestNormalizeURL(t *testing.T) {
 		{"HTTPS://Example.COM/path", "https://example.com/path"},
 		{"http://example.com/", "http://example.com"},
 		{"http://example.com/page", "http://example.com/page"},
+		{"https://u:p@Example.com/app?x=1", "https://example.com/app?x=1"},
+		{"https://user@example.com/", "https://example.com"},
+		{"https://u:p@example.com:8443/users/@alice?x=1", "https://example.com:8443/users/@alice?x=1"},
+		{"https://example.com/@me?r=http://a:b@x/", "https://example.com/@me?r=http://a:b@x/"},
 	}
 	for _, tc := range tests {
 		got := normalizeURL(tc.raw)
@@ -902,6 +906,139 @@ func TestNormalizeURL(t *testing.T) {
 	}
 }

+// storedURLRow reads back the url, domain and port of one asset row.
+func storedURLRow(t *testing.T, d *DB, id int64) (string, string, int) {
+	t.Helper()
+	var u, domain *string
+	var port *int
+	if err := d.QueryRow(`SELECT url, domain, port FROM assets WHERE id = $1`, id).Scan(&u, &domain, &port); err != nil {
+		t.Fatalf("read asset %d: %v", id, err)
+	}
+	var us, ds string
+	var p int
+	if u != nil {
+		us = *u
+	}
+	if domain != nil {
+		ds = *domain
+	}
+	if port != nil {
+		p = *port
+	}
+	return us, ds, p
+}
+
+func TestUpsertHTTPServiceStripsUserinfo(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	cleanup := func() {
+		d.Exec(`DELETE FROM assets WHERE url LIKE '%storeuserinfo-svc.com%' OR domain LIKE '%storeuserinfo-svc.com' OR root_domain = 'storeuserinfo-svc.com'`)
+	}
+	cleanup()
+	defer cleanup()
+
+	id1, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: "https://alice:s3cr3t-one@App.storeuserinfo-svc.com/app?x=1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	u, domain, port := storedURLRow(t, d, id1)
+	if u != "https://app.storeuserinfo-svc.com/app?x=1" || domain != "app.storeuserinfo-svc.com" || port != 443 {
+		t.Fatalf("stored url=%q domain=%q port=%d", u, domain, port)
+	}
+
+	// Same URL, different credentials: deduplicated onto the same row.
+	id2, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: "https://bob:s3cr3t-two@app.storeuserinfo-svc.com/app?x=1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if id2 != id1 {
+		t.Errorf("dedup across credentials failed: %d vs %d", id2, id1)
+	}
+	var n int
+	d.QueryRow(`SELECT count(*) FROM assets WHERE type = 'service' AND service_type = 'http' AND url LIKE '%storeuserinfo-svc.com/app%'`).Scan(&n)
+	if n != 1 {
+		t.Errorf("want 1 http service row, got %d", n)
+	}
+	d.QueryRow(`SELECT count(*) FROM assets WHERE url LIKE '%s3cr3t%' AND url LIKE '%storeuserinfo-svc.com%'`).Scan(&n)
+	if n != 0 {
+		t.Errorf("credentials stored in %d asset url(s)", n)
+	}
+
+	// @ in the path is kept; only the userinfo goes.
+	id3, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: "https://u:p@app.storeuserinfo-svc.com:8443/users/@alice?x=1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	u, domain, port = storedURLRow(t, d, id3)
+	if u != "https://app.storeuserinfo-svc.com:8443/users/@alice?x=1" || domain != "app.storeuserinfo-svc.com" || port != 8443 {
+		t.Errorf("@ path: stored url=%q domain=%q port=%d", u, domain, port)
+	}
+
+	// Regression: a URL without userinfo is stored unchanged.
+	plain := "https://plain.storeuserinfo-svc.com/@me?q=1"
+	id4, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: plain})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if u, _, _ = storedURLRow(t, d, id4); u != plain {
+		t.Errorf("plain url: want %q, got %q", plain, u)
+	}
+}
+
+func TestUpsertEndpointStripsUserinfo(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	cleanup := func() {
+		d.Exec(`DELETE FROM assets WHERE url LIKE '%storeuserinfo-ep.com%' OR domain LIKE '%storeuserinfo-ep.com' OR root_domain = 'storeuserinfo-ep.com'`)
+	}
+	cleanup()
+	defer cleanup()
+
+	id1, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: "https://alice:s3cr3t-one@api.storeuserinfo-ep.com/users?id=1", Method: "GET"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	u, domain, port := storedURLRow(t, d, id1)
+	if u != "https://api.storeuserinfo-ep.com/users?id=1" || domain != "api.storeuserinfo-ep.com" || port != 443 {
+		t.Fatalf("stored url=%q domain=%q port=%d", u, domain, port)
+	}
+
+	id2, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: "https://bob:s3cr3t-two@api.storeuserinfo-ep.com/users?id=1", Method: "GET"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if id2 != id1 {
+		t.Errorf("endpoint dedup across credentials failed: %d vs %d", id2, id1)
+	}
+	var n int
+	d.QueryRow(`SELECT count(*) FROM assets WHERE type = 'endpoint' AND url LIKE '%storeuserinfo-ep.com/users%'`).Scan(&n)
+	if n != 1 {
+		t.Errorf("want 1 endpoint row, got %d", n)
+	}
+	d.QueryRow(`SELECT count(*) FROM assets WHERE url LIKE '%s3cr3t%' AND url LIKE '%storeuserinfo-ep.com%'`).Scan(&n)
+	if n != 0 {
+		t.Errorf("credentials stored in %d asset url(s)", n)
+	}
+
+	id3, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: "http://u:p@api.storeuserinfo-ep.com/users/@alice?x=1", Method: "POST"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	u, _, port = storedURLRow(t, d, id3)
+	if u != "http://api.storeuserinfo-ep.com/users/@alice?x=1" || port != 80 {
+		t.Errorf("@ path: stored url=%q port=%d", u, port)
+	}
+
+	plain := "https://plain.storeuserinfo-ep.com/@me?q=1"
+	id4, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: plain, Method: "GET"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if u, _, _ = storedURLRow(t, d, id4); u != plain {
+		t.Errorf("plain url: want %q, got %q", plain, u)
+	}
+}
+
 func TestAssetPaginationUsesStableIDTieBreaker(t *testing.T) {
 	d, assets, _ := testSetup(t)
 	defer d.Close()
@@ -1010,7 +1147,8 @@ func TestAssetIPRejectsHostname(t *testing.T) {
 // are excluded. Direct source tasks' scope is included.
 func TestQueryDSLInScopeMembership(t *testing.T) {
 	d, assets, _ := testSetup(t)
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	task, err := d.CreateTask("scope membership", "goal", nil, 0, 0)
 	if err != nil {
@@ -1020,7 +1158,10 @@ func TestQueryDSLInScopeMembership(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID); _ = d.DeleteTask(src.ID) })
+	t.Cleanup(func() {
+		cleanupErr(t, "task", d.DeleteTask(task.ID))
+		cleanupErr(t, "source task", d.DeleteTask(src.ID))
+	})
 	// task ← src is a direct source relation.
 	if _, err := d.Exec(`INSERT INTO task_relations(task_id, source_task_id) VALUES ($1,$2)`, task.ID, src.ID); err != nil {
 		t.Fatal(err)
@@ -1079,7 +1220,7 @@ func TestQueryDSLInScopeMembership(t *testing.T) {
 		}
 	}
 	t.Cleanup(func() {
-		_, _ = d.Exec(`DELETE FROM assets WHERE id=ANY($1::bigint[]) OR id=ANY($2::bigint[])`, inScopeIDs, outIDs)
+		cleanupExec(t, d, `DELETE FROM assets WHERE id=ANY($1::bigint[]) OR id=ANY($2::bigint[])`, inScopeIDs, outIDs)
 	})

 	got, err := assets.QueryDSLInScope(marker, "", task.ID, 50, 0)
@@ -0,0 +1,34 @@
+package db
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+// These helpers report a failed test cleanup instead of swallowing it: a silent
+// failure leaves the test's rows behind for the next run on the same database.
+
+// cleanupErr reports err from the cleanup step described by what.
+func cleanupErr(t *testing.T, what string, err error) {
+	t.Helper()
+	if err != nil {
+		t.Errorf("cleanup %s: %v", what, err)
+	}
+}
+
+// cleanupExec runs one cleanup statement. Matching no row is not an error.
+func cleanupExec(t *testing.T, d *DB, query string, args ...any) {
+	t.Helper()
+	_, err := d.Exec(query, args...)
+	cleanupErr(t, query, err)
+}
+
+// cleanupProfile deletes a profile through DeleteProfile. A profile the test
+// already deleted is not an error.
+func cleanupProfile(t *testing.T, d *DB, id int64) {
+	t.Helper()
+	if err := d.DeleteProfile(id); !errors.Is(err, ErrLLMProfileNotFound) {
+		cleanupErr(t, fmt.Sprintf("profile %d", id), err)
+	}
+}
@@ -153,57 +153,13 @@ type LLMRecord struct {
 	RawResponse string `json:"raw_response,omitempty"`
 }

-const llmRecordsSchema = `
-CREATE TABLE IF NOT EXISTS llm_records (
-    id            BIGSERIAL PRIMARY KEY,
-    ts            TIMESTAMPTZ DEFAULT now(),
-    model         TEXT,
-    profile_name  TEXT,
-    session_id    TEXT,
-    task_id       TEXT,
-    worker        TEXT,
-    latency_ms    INTEGER,
-    input_tokens  INTEGER,
-    output_tokens INTEGER,
-    cache_read    INTEGER,
-    cache_write   INTEGER,
-    status        TEXT,
-    error         TEXT,
-    request_body  TEXT,
-    response_body TEXT,
-    raw_request   TEXT,
-    raw_response  TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_llm_records_ts ON llm_records(ts);
-CREATE INDEX IF NOT EXISTS idx_llm_records_session ON llm_records(session_id);
-`
-
-// llmRecordsMigrate adds new columns to existing tables.
-const llmRecordsMigrate = `
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS task_id TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS worker TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS profile_name TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_request TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_response TEXT;
-`
-
-// EnsureLLMRecordsTable creates the llm_records table if it does not exist.
+// EnsureLLMRecordsTable creates the llm_records table if it does not exist, by
+// re-running the idempotent migration 0001_llm_records.sql. Open already applies
+// it through the migration runner; this stays for callers that want the table
+// guaranteed. It takes the schema lock in its own transaction, so it must not be
+// called while Open holds that lock.
 func (d *DB) EnsureLLMRecordsTable() error {
-	tx, err := d.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback() //nolint:errcheck
-	if err := coordinateWithSchemaMigration(tx); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmRecordsSchema); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
-		return err
-	}
-	return tx.Commit()
+	return d.execEmbeddedMigration(llmRecordsMigration)
 }

 // InsertLLMRecord stores one LLM call record.
@@ -10,7 +10,8 @@ import (

 func TestAssetUpsertWaitsForCompanyScopeMutation(t *testing.T) {
 	d, assets, companies := testSetup(t)
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	domain := fmt.Sprintf("scope-lock-%d.example", stamp)
@@ -23,8 +24,8 @@ func TestAssetUpsertWaitsForCompanyScopeMutation(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_, _ = d.Exec(`DELETE FROM assets WHERE domain = $1`, domain)
-		_, _ = d.Exec(`DELETE FROM companies WHERE id IN ($1,$2)`, oldCompany, newCompany)
+		cleanupExec(t, d, `DELETE FROM assets WHERE domain = $1`, domain)
+		cleanupExec(t, d, `DELETE FROM companies WHERE id IN ($1,$2)`, oldCompany, newCompany)
 	})
 	if added, _, invalid, validationErrors, err := companies.AddScopeInputsChecked(oldCompany, []ScopeInput{
 		{Kind: "domain", Value: domain},
@@ -90,7 +91,8 @@ func TestAssetUpsertWaitsForCompanyScopeMutation(t *testing.T) {

 func TestResolveAndRecomputeUseStableCompanyIDTieBreak(t *testing.T) {
 	d, assets, companies := testSetup(t)
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	first, _, err := companies.UpsertCompany(fmt.Sprintf("Tie Break First %d", stamp), "")
@@ -109,8 +111,8 @@ func TestResolveAndRecomputeUseStableCompanyIDTieBreak(t *testing.T) {
 	icp := fmt.Sprintf("ICP-TIE-%d", stamp)
 	appName := fmt.Sprintf("Tie Break App %d", stamp)
 	t.Cleanup(func() {
-		_, _ = d.Exec(`DELETE FROM assets WHERE domain=$1 OR ip=$2 OR app_name=$3`, domain, ip, appName)
-		_, _ = d.Exec(`DELETE FROM companies WHERE id IN ($1,$2)`, first, second)
+		cleanupExec(t, d, `DELETE FROM assets WHERE domain=$1 OR ip=$2 OR app_name=$3`, domain, ip, appName)
+		cleanupExec(t, d, `DELETE FROM companies WHERE id IN ($1,$2)`, first, second)
 	})

 	rules := []ScopeInput{
@@ -11,7 +11,8 @@ func TestConversationPinOrderingAndPatch(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	first, err := d.CreateConversation("mainagent", fmt.Sprintf("pin-first-%d", suffix), nil)
@@ -24,8 +25,8 @@ func TestConversationPinOrderingAndPatch(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteConversation(first.ID)
-		_ = d.DeleteConversation(second.ID)
+		cleanupErr(t, "first conversation", d.DeleteConversation(first.ID))
+		cleanupErr(t, "second conversation", d.DeleteConversation(second.ID))
 	})

 	pinned := true
@@ -102,12 +102,13 @@ func TestRecordCoverageUpsertHistory(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })
 	task, err := d.CreateTaskWithOptions("coverage ledger", "g", TaskCreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	as := d.Assets()
 	store := d.Exploration(task.ExplorationID)
 	for _, vc := range []string{"SQLi", "XSS"} {
@@ -158,12 +159,13 @@ func TestCoverageCompletenessBudgetThenReviveDone(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })
 	task, err := d.CreateTaskWithOptions("coverage revive", "g", TaskCreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	as, store := d.Assets(), d.Exploration(task.ExplorationID)
 	state := func() string {
 		t.Helper()
@@ -221,12 +223,13 @@ func TestRecordCoverageRowCap(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })
 	task, err := d.CreateTaskWithOptions("coverage cap", "g", TaskCreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	if _, err := d.Exec(`INSERT INTO coverage_ledger(task_id, surface, risk_area, outcome)
 SELECT $1, '/s' || g, 'xss', 'needs_follow_up' FROM generate_series(1, $2::int) g`, task.ID, MaxCoverageRowsPerTask); err != nil {
 		t.Fatal(err)
@@ -12,7 +12,8 @@ func TestCustomToolCRUD(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	key := "ct_test_tool"
 	_ = d.DeleteCustomTool(key) // clean slate
@@ -30,7 +31,7 @@ func TestCustomToolCRUD(t *testing.T) {
 	if err := d.CreateCustomTool(in); err != nil {
 		t.Fatalf("create: %v", err)
 	}
-	t.Cleanup(func() { _ = d.DeleteCustomTool(key) })
+	t.Cleanup(func() { cleanupErr(t, "custom tool", d.DeleteCustomTool(key)) })

 	got, err := d.GetTool(key)
 	if err != nil || got == nil {
@@ -1,6 +1,15 @@
 // Package db is the PostgreSQL data source for ARTEX (replacing the old graph
 // single-file SQLite). It opens the connection, applies the schema, and seeds
 // the built-in agents and variable catalog.
+//
+// Schema changes:
+//   - schema.sql remains the idempotent base, executed on every startup.
+//   - Any new DDL goes in a numbered migration, migrations/NNNN_slug.sql.
+//     Open applies pending migrations once, in version order, and records
+//     each in schema_migrations with the SHA-256 of its file.
+//   - A published migration is never edited: Open refuses to start when an
+//     applied migration's checksum no longer matches its file. Fix forward
+//     with a new migration instead.
 package db

 import (
@@ -9,6 +18,7 @@ import (
 	_ "embed"
 	"errors"
 	"fmt"
+	"io/fs"
 	"net/url"
 	"os"
 	"strings"
@@ -153,8 +163,15 @@ func ensureDatabase(dsn string) error {
 	return nil
 }

-// Open connects, applies the schema (idempotent), and seeds builtin rows.
+// Open connects, applies the schema (idempotent) and the embedded versioned
+// migrations, and seeds builtin rows.
 func Open(dsn string) (*DB, error) {
+	return openWithMigrations(dsn, migrationsFS())
+}
+
+// openWithMigrations is Open with the migration set as a parameter, so tests
+// can pass an fstest.MapFS.
+func openWithMigrations(dsn string, migrations fs.FS) (*DB, error) {
 	if err := ensureDatabase(dsn); err != nil {
 		return nil, err
 	}
@@ -187,6 +204,11 @@ func Open(dsn string) (*DB, error) {
 		if err := applySchemaWithRetry(context.Background(), conn, time.Sleep); err != nil {
 			return fmt.Errorf("apply schema: %w", err)
 		}
+		// Runs on conn, the lock holder; it must not call EnsureLLM*Table,
+		// whose own lock request would wait on conn forever.
+		if err := applyMigrations(context.Background(), conn, migrations); err != nil {
+			return fmt.Errorf("apply migrations: %w", err)
+		}
 		if err := d.seedBuiltins(); err != nil {
 			return fmt.Errorf("seed builtins: %w", err)
 		}
@@ -10,13 +10,14 @@ func TestInheritedActivityReadsRequireTerminalIntent(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	expID, err := d.CreateExploration("source activity boundary", "source activity boundary")
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.Exec(`DELETE FROM explorations WHERE id=$1`, expID) })
+	t.Cleanup(func() { cleanupExec(t, d, `DELETE FROM explorations WHERE id=$1`, expID) })
 	store := d.Exploration(expID)
 	intentID, err := store.AddIntent(map[string]any{"summary": "source work"}, 1, nil, "planner")
 	if err != nil {
@@ -93,7 +94,8 @@ func TestExplorationDirectSourceReadView(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	grand, err := d.CreateTask("grand source", "grand goal", nil, 0, 0)
 	if err != nil {
@@ -112,9 +114,9 @@ func TestExplorationDirectSourceReadView(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(current.ID)
-		_ = d.DeleteTask(source.ID)
-		_ = d.DeleteTask(grand.ID)
+		cleanupErr(t, "current task", d.DeleteTask(current.ID))
+		cleanupErr(t, "source task", d.DeleteTask(source.ID))
+		cleanupErr(t, "grand task", d.DeleteTask(grand.ID))
 	})

 	grandStore := d.Exploration(grand.ExplorationID)
@@ -315,7 +317,8 @@ func TestTaskAssetContextUsesDirectSourceScopeAndAnchors(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	grand, err := d.CreateTask("grand assets", "grand goal", nil, 0, 0)
 	if err != nil {
@@ -351,10 +354,11 @@ func TestTaskAssetContextUsesDirectSourceScopeAndAnchors(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(current.ID)
-		_ = d.DeleteTask(source.ID)
-		_ = d.DeleteTask(grand.ID)
-		_, _ = assets.DeleteByIDs([]int64{sourceTestedID, sourceUntestedID, sourceAnchoredOnlyID, grandID})
+		cleanupErr(t, "current task", d.DeleteTask(current.ID))
+		cleanupErr(t, "source task", d.DeleteTask(source.ID))
+		cleanupErr(t, "grand task", d.DeleteTask(grand.ID))
+		_, err := assets.DeleteByIDs([]int64{sourceTestedID, sourceUntestedID, sourceAnchoredOnlyID, grandID})
+		cleanupErr(t, "assets", err)
 	})
 	if _, err := assets.AddAgentScope(source.ID, "root_domain", sourceTestedDomain, "test", "agent"); err != nil {
 		t.Fatal(err)
@@ -1,11 +1,12 @@
 package db

 import (
+	"io/fs"
 	"testing"
 )

 // The raw-body columns were added after release, so existing installs only get
-// them through llmRecordsMigrate. This runs the real migration against the dev
+// them through the ALTERs in 0001_llm_records.sql. This runs the real migration against the dev
 // PG on a table stripped back to its pre-upgrade shape, inside a transaction
 // that is always rolled back — PG does transactional DDL, so nothing persists.
 func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
@@ -17,6 +18,11 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	if err := d.EnsureLLMRecordsTable(); err != nil {
 		t.Fatalf("ensure: %v", err)
 	}
+	content, err := fs.ReadFile(migrationsFS(), llmRecordsMigration)
+	if err != nil {
+		t.Fatal(err)
+	}
+	llmRecordsMigrateSQL := string(content)

 	tx, err := d.Begin()
 	if err != nil {
@@ -28,7 +34,7 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	if _, err := tx.Exec(`ALTER TABLE llm_records DROP COLUMN IF EXISTS raw_request, DROP COLUMN IF EXISTS raw_response`); err != nil {
 		t.Fatalf("simulate old table: %v", err)
 	}
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
+	if _, err := tx.Exec(llmRecordsMigrateSQL); err != nil {
 		t.Fatalf("migrate: %v", err)
 	}

@@ -45,7 +51,7 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	}

 	// Re-running must stay a no-op (the migration runs on every startup).
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
+	if _, err := tx.Exec(llmRecordsMigrateSQL); err != nil {
 		t.Fatalf("migrate is not idempotent: %v", err)
 	}

@@ -26,41 +26,11 @@ type LLMUsage struct {
 	Status        string `json:"status"` // ok | error
 }

-const llmUsageSchema = `
-CREATE TABLE IF NOT EXISTS llm_usage (
-    id             BIGSERIAL PRIMARY KEY,
-    ts             TIMESTAMPTZ NOT NULL DEFAULT now(),
-    task_id        TEXT,
-    exploration_id BIGINT,
-    worker         TEXT,
-    model          TEXT,
-    profile_name   TEXT,
-    latency_ms     INTEGER,
-    input_tokens   INTEGER NOT NULL DEFAULT 0,
-    output_tokens  INTEGER NOT NULL DEFAULT 0,
-    cache_read     INTEGER NOT NULL DEFAULT 0,
-    cache_write    INTEGER NOT NULL DEFAULT 0,
-    status         TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_task  ON llm_usage(task_id);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_model ON llm_usage(task_id, model);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_exp   ON llm_usage(exploration_id);
-`
-
-// EnsureLLMUsageTable creates the llm_usage metering table if it does not exist.
+// EnsureLLMUsageTable creates the llm_usage metering table if it does not exist,
+// by re-running the idempotent migration 0002_llm_usage.sql (see
+// EnsureLLMRecordsTable for the locking rule).
 func (d *DB) EnsureLLMUsageTable() error {
-	tx, err := d.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback() //nolint:errcheck
-	if err := coordinateWithSchemaMigration(tx); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmUsageSchema); err != nil {
-		return err
-	}
-	return tx.Commit()
+	return d.execEmbeddedMigration(llmUsageMigration)
 }

 // InsertLLMUsage appends one metering row. Best-effort: callers log and continue on
@@ -0,0 +1,217 @@
+package db
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"embed"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"path"
+	"regexp"
+	"sort"
+	"strconv"
+)
+
+// Versioned migrations, applied by Open after schema.sql (rules in the package
+// comment, db.go).
+//
+//go:embed migrations/*.sql
+var embeddedMigrationFiles embed.FS
+
+// Embedded migration files that EnsureLLMRecordsTable / EnsureLLMUsageTable
+// re-run on demand.
+const (
+	llmRecordsMigration = "0001_llm_records.sql"
+	llmUsageMigration   = "0002_llm_usage.sql"
+)
+
+// migrationsFS is the embedded migrations directory, rooted so file names are
+// the bare NNNN_slug.sql names.
+func migrationsFS() fs.FS {
+	sub, err := fs.Sub(embeddedMigrationFiles, "migrations")
+	if err != nil {
+		panic(err) // the embed pattern guarantees the directory exists
+	}
+	return sub
+}
+
+const schemaMigrationsTable = `
+CREATE TABLE IF NOT EXISTS schema_migrations (
+    version    INTEGER PRIMARY KEY,
+    name       TEXT,
+    checksum   TEXT,
+    applied_at TIMESTAMPTZ DEFAULT now()
+);
+`
+
+var migrationNameRE = regexp.MustCompile(`^([0-9]{4})_[a-z0-9_]+\.sql$`)
+
+type migration struct {
+	version  int
+	name     string // file name, e.g. 0001_llm_records.sql
+	sql      string
+	checksum string // hex SHA-256 of the file content
+}
+
+// loadMigrations reads every *.sql file at the root of fsys, sorted by version.
+// A .sql file that does not follow NNNN_slug.sql, or a duplicated version, is
+// an error so a typo cannot silently skip a migration.
+func loadMigrations(fsys fs.FS) ([]migration, error) {
+	entries, err := fs.ReadDir(fsys, ".")
+	if err != nil {
+		return nil, fmt.Errorf("read migrations: %w", err)
+	}
+	var out []migration
+	seen := map[int]string{}
+	for _, e := range entries {
+		if e.IsDir() || path.Ext(e.Name()) != ".sql" {
+			continue
+		}
+		m := migrationNameRE.FindStringSubmatch(e.Name())
+		if m == nil {
+			return nil, fmt.Errorf("migration file %s: name must be NNNN_slug.sql", e.Name())
+		}
+		version, _ := strconv.Atoi(m[1])
+		if version == 0 {
+			return nil, fmt.Errorf("migration file %s: version must be > 0", e.Name())
+		}
+		if prev, dup := seen[version]; dup {
+			return nil, fmt.Errorf("migration version %04d is used by both %s and %s", version, prev, e.Name())
+		}
+		seen[version] = e.Name()
+		content, err := fs.ReadFile(fsys, e.Name())
+		if err != nil {
+			return nil, fmt.Errorf("read migration %s: %w", e.Name(), err)
+		}
+		sum := sha256.Sum256(content)
+		out = append(out, migration{version: version, name: e.Name(), sql: string(content), checksum: hex.EncodeToString(sum[:])})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].version < out[j].version })
+	return out, nil
+}
+
+// applyMigrations brings the database up to the migrations in fsys. conn must
+// be the connection that holds the schema advisory lock (see Open); every
+// statement runs on it, so nothing here waits on that lock again.
+//
+// It fails closed before applying anything when the database records a version
+// this binary does not ship, or an applied migration whose name or checksum
+// differs from the file, or a pending migration is older than the newest
+// applied one. Each pending migration then runs in its own
+// transaction together with its schema_migrations row; a failure rolls that one
+// back and stops, leaving earlier migrations applied.
+func applyMigrations(ctx context.Context, conn *sql.Conn, fsys fs.FS) error {
+	migrations, err := loadMigrations(fsys)
+	if err != nil {
+		return err
+	}
+	if _, err := conn.ExecContext(ctx, schemaMigrationsTable); err != nil {
+		return fmt.Errorf("create schema_migrations: %w", err)
+	}
+
+	type appliedRow struct{ name, checksum string }
+	applied := map[int]appliedRow{}
+	rows, err := conn.QueryContext(ctx, `SELECT version, COALESCE(name,''), COALESCE(checksum,'') FROM schema_migrations ORDER BY version`)
+	if err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+	var appliedVersions []int
+	for rows.Next() {
+		var v int
+		var r appliedRow
+		if err := rows.Scan(&v, &r.name, &r.checksum); err != nil {
+			rows.Close()
+			return fmt.Errorf("read schema_migrations: %w", err)
+		}
+		applied[v] = r
+		appliedVersions = append(appliedVersions, v)
+	}
+	if err := rows.Close(); err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+
+	byVersion := map[int]migration{}
+	for _, m := range migrations {
+		byVersion[m.version] = m
+	}
+	for _, v := range appliedVersions {
+		r := applied[v]
+		m, ok := byVersion[v]
+		if !ok {
+			return fmt.Errorf("migration %04d (%s) is applied in the database but missing from this binary", v, r.name)
+		}
+		if r.name != m.name {
+			return fmt.Errorf("migration %04d: database records file %s but this binary has %s", v, r.name, m.name)
+		}
+		if r.checksum != m.checksum {
+			return fmt.Errorf("migration %04d (%s): checksum changed since it was applied (database %s, file %s); published migrations must not be edited", v, m.name, r.checksum, m.checksum)
+		}
+	}
+
+	// A pending version below the newest applied one would run out of order on
+	// this database: refuse it before writing anything.
+	if len(appliedVersions) > 0 {
+		maxVersion := appliedVersions[len(appliedVersions)-1]
+		for _, m := range migrations {
+			if _, done := applied[m.version]; !done && m.version < maxVersion {
+				return fmt.Errorf("migration %s (%04d) is older than the applied migration %s (%04d); migrations must be appended, never inserted",
+					m.name, m.version, applied[maxVersion].name, maxVersion)
+			}
+		}
+	}
+
+	for _, m := range migrations {
+		if _, done := applied[m.version]; done {
+			continue
+		}
+		if err := applyMigration(ctx, conn, m); err != nil {
+			return fmt.Errorf("migration %04d (%s): %w", m.version, m.name, err)
+		}
+	}
+	return nil
+}
+
+func applyMigration(ctx context.Context, conn *sql.Conn, m migration) error {
+	tx, err := conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck // no-op after Commit
+	// No args: pgx sends it over the simple protocol, which allows several statements.
+	if _, err := tx.ExecContext(ctx, m.sql); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migrations(version, name, checksum) VALUES ($1, $2, $3)`, m.version, m.name, m.checksum); err != nil {
+		return fmt.Errorf("record: %w", err)
+	}
+	return tx.Commit()
+}
+
+// execEmbeddedMigration re-runs one embedded (idempotent) migration file in its
+// own transaction under the schema lock. It does not touch schema_migrations.
+// The lock is taken on a pooled connection, so this must never be called from
+// inside Open's withSchemaMigrationLock callback: it would wait on the lock
+// that callback's connection holds.
+func (d *DB) execEmbeddedMigration(name string) error {
+	content, err := fs.ReadFile(migrationsFS(), name)
+	if err != nil {
+		return err
+	}
+	tx, err := d.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck
+	if err := coordinateWithSchemaMigration(tx); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(string(content)); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
@@ -0,0 +1,415 @@
+package db
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"io/fs"
+	"net/url"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+)
+
+func TestLoadMigrationsValidation(t *testing.T) {
+	ms, err := loadMigrations(fstest.MapFS{
+		"0010_b.sql": {Data: []byte("SELECT 10;")},
+		"0002_a.sql": {Data: []byte("SELECT 2;")},
+		"README.md":  {Data: []byte("not a migration")},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(ms) != 2 || ms[0].version != 2 || ms[1].version != 10 {
+		t.Fatalf("want versions [2 10], got %+v", ms)
+	}
+	sum := sha256.Sum256([]byte("SELECT 2;"))
+	if ms[0].checksum != hex.EncodeToString(sum[:]) || ms[0].name != "0002_a.sql" {
+		t.Fatalf("migration 0002 = %+v", ms[0])
+	}
+
+	for name, fsys := range map[string]fstest.MapFS{
+		"bad name":          {"2_a.sql": {Data: []byte("SELECT 1;")}},
+		"duplicate version": {"0001_a.sql": {Data: []byte("SELECT 1;")}, "0001_b.sql": {Data: []byte("SELECT 1;")}},
+	} {
+		if _, err := loadMigrations(fsys); err == nil {
+			t.Errorf("%s: want error", name)
+		}
+	}
+}
+
+// scratchDSN returns the DSN of a new, empty database named after the test
+// database plus suffix, dropped again when the test ends.
+func scratchDSN(t *testing.T, suffix string) string {
+	t.Helper()
+	u, err := url.Parse(testDSN(t))
+	if err != nil || strings.Trim(u.Path, "/") == "" {
+		t.Skipf("DSN has no database name — skipping")
+	}
+	name := strings.TrimPrefix(u.Path, "/") + "_mig_" + suffix
+	admin := *u
+	admin.Path = "/postgres"
+	conn, err := sql.Open("pgx", admin.String())
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	if err := conn.Ping(); err != nil {
+		conn.Close()
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	drop := func() error {
+		_, err := conn.Exec(`DROP DATABASE IF EXISTS "` + name + `" WITH (FORCE)`)
+		return err
+	}
+	if err := drop(); err != nil {
+		t.Fatalf("drop leftover %s: %v", name, err)
+	}
+	t.Cleanup(func() {
+		if err := drop(); err != nil {
+			t.Errorf("drop %s: %v", name, err)
+		}
+		conn.Close()
+	})
+	u.Path = "/" + name
+	return u.String()
+}
+
+// withEmbedded returns the embedded migrations plus extra files.
+func withEmbedded(t *testing.T, extra fstest.MapFS) fstest.MapFS {
+	t.Helper()
+	out := fstest.MapFS{}
+	for _, name := range []string{llmRecordsMigration, llmUsageMigration} {
+		content, err := fs.ReadFile(migrationsFS(), name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		out[name] = &fstest.MapFile{Data: content}
+	}
+	for name, f := range extra {
+		out[name] = f
+	}
+	return out
+}
+
+type appliedMigration struct {
+	version        int
+	name, checksum string
+	appliedAt      time.Time
+}
+
+func readApplied(t *testing.T, d *DB) []appliedMigration {
+	t.Helper()
+	rows, err := d.Query(`SELECT version, name, checksum, applied_at FROM schema_migrations ORDER BY version`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var out []appliedMigration
+	for rows.Next() {
+		var m appliedMigration
+		if err := rows.Scan(&m.version, &m.name, &m.checksum, &m.appliedAt); err != nil {
+			t.Fatal(err)
+		}
+		out = append(out, m)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+
+// assertEmbeddedApplied checks that both LLM tables exist and that 0001 and
+// 0002 are recorded with the SHA-256 of their embedded files.
+func assertEmbeddedApplied(t *testing.T, d *DB) []appliedMigration {
+	t.Helper()
+	for _, table := range []string{"llm_records", "llm_usage"} {
+		var exists bool
+		if err := d.QueryRow(`SELECT to_regclass($1) IS NOT NULL`, table).Scan(&exists); err != nil {
+			t.Fatal(err)
+		}
+		if !exists {
+			t.Errorf("table %s missing after Open", table)
+		}
+	}
+	applied := readApplied(t, d)
+	want := []struct {
+		version int
+		name    string
+	}{{1, llmRecordsMigration}, {2, llmUsageMigration}}
+	if len(applied) != len(want) {
+		t.Fatalf("schema_migrations = %+v, want versions 1 and 2", applied)
+	}
+	for i, w := range want {
+		content, err := fs.ReadFile(migrationsFS(), w.name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		sum := sha256.Sum256(content)
+		got := applied[i]
+		if got.version != w.version || got.name != w.name || got.checksum != hex.EncodeToString(sum[:]) {
+			t.Errorf("schema_migrations row %d = %+v, want version %d name %s checksum %x", i, got, w.version, w.name, sum)
+		}
+	}
+	return applied
+}
+
+// assertReopenIsNoop opens dsn again and checks that no migration row was added
+// or re-applied, then that both Ensure functions stay idempotent.
+func assertReopenIsNoop(t *testing.T, dsn string, before []appliedMigration) {
+	t.Helper()
+	d, err := Open(dsn)
+	if err != nil {
+		t.Fatalf("second Open: %v", err)
+	}
+	defer d.Close()
+	after := assertEmbeddedApplied(t, d)
+	for i := range before {
+		if !after[i].appliedAt.Equal(before[i].appliedAt) {
+			t.Errorf("migration %04d applied_at changed: %v -> %v", before[i].version, before[i].appliedAt, after[i].appliedAt)
+		}
+	}
+	for range 2 {
+		if err := d.EnsureLLMRecordsTable(); err != nil {
+			t.Fatalf("EnsureLLMRecordsTable after Open: %v", err)
+		}
+		if err := d.EnsureLLMUsageTable(); err != nil {
+			t.Fatalf("EnsureLLMUsageTable after Open: %v", err)
+		}
+	}
+	if got := readApplied(t, d); len(got) != len(before) {
+		t.Errorf("Ensure functions changed schema_migrations: %+v", got)
+	}
+}
+
+// TestOpenRecordsEmbeddedMigrations runs against the package test database:
+// fresh on a first run, already migrated on a second run.
+func TestOpenRecordsEmbeddedMigrations(t *testing.T) {
+	dsn := testDSN(t)
+	d, err := Open(dsn)
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+func TestOpenMigratesFreshDatabase(t *testing.T) {
+	dsn := scratchDSN(t, "fresh")
+	d, err := Open(dsn)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+// An install that predates schema_migrations already has both tables (created
+// by the Ensure functions, possibly without the later llm_records columns).
+// Open must adopt them: run 0001/0002 over them and record both.
+func TestOpenAdoptsExistingTablesWithoutSchemaMigrations(t *testing.T) {
+	dsn := scratchDSN(t, "legacy")
+	d, err := openWithMigrations(dsn, fstest.MapFS{})
+	if err != nil {
+		t.Fatalf("Open without migrations: %v", err)
+	}
+	if err := d.EnsureLLMRecordsTable(); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.EnsureLLMUsageTable(); err != nil {
+		t.Fatal(err)
+	}
+	for _, q := range []string{
+		`ALTER TABLE llm_records DROP COLUMN raw_request, DROP COLUMN raw_response`,
+		`INSERT INTO llm_records(model) VALUES ('legacy')`,
+		`INSERT INTO llm_usage(model, status) VALUES ('legacy', 'ok')`,
+		`DROP TABLE schema_migrations`,
+	} {
+		if _, err := d.Exec(q); err != nil {
+			t.Fatalf("%s: %v", q, err)
+		}
+	}
+	d.Close()
+
+	d, err = Open(dsn)
+	if err != nil {
+		t.Fatalf("Open on existing tables: %v", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	var records, usage, rawCols int
+	if err := d.QueryRow(`SELECT count(*) FROM llm_records WHERE model='legacy'`).Scan(&records); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.QueryRow(`SELECT count(*) FROM llm_usage WHERE model='legacy'`).Scan(&usage); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.QueryRow(`SELECT count(*) FROM information_schema.columns
+		WHERE table_name='llm_records' AND column_name IN ('raw_request','raw_response')`).Scan(&rawCols); err != nil {
+		t.Fatal(err)
+	}
+	if records != 1 || usage != 1 || rawCols != 2 {
+		t.Errorf("after adopt: llm_records legacy rows=%d llm_usage legacy rows=%d raw columns=%d, want 1 1 2", records, usage, rawCols)
+	}
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+// probeMigration records how many migrations were already applied when it
+// ran, and fails if it ever runs twice (CREATE TABLE without IF NOT EXISTS).
+var probeMigration = &fstest.MapFile{Data: []byte(`
+CREATE TABLE migration_probe (applied_before INTEGER);
+INSERT INTO migration_probe SELECT count(*) FROM schema_migrations;
+`)}
+
+func TestMigrationsApplyNewVersionOnceInOrder(t *testing.T) {
+	dsn := scratchDSN(t, "order")
+	fsys := withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration})
+	for pass := 1; pass <= 2; pass++ {
+		d, err := openWithMigrations(dsn, fsys)
+		if err != nil {
+			t.Fatalf("pass %d: %v", pass, err)
+		}
+		applied := readApplied(t, d)
+		if len(applied) != 3 || applied[2].version != 3 || applied[2].name != "0003_probe.sql" {
+			t.Fatalf("pass %d: schema_migrations = %+v", pass, applied)
+		}
+		var runs, before int
+		if err := d.QueryRow(`SELECT count(*), max(applied_before) FROM migration_probe`).Scan(&runs, &before); err != nil {
+			t.Fatal(err)
+		}
+		if runs != 1 || before != 2 {
+			t.Fatalf("pass %d: probe ran %d times with %d migrations before it, want 1 and 2", pass, runs, before)
+		}
+		d.Close()
+	}
+}
+
+func TestMigrationsFailClosed(t *testing.T) {
+	dsn := scratchDSN(t, "closed")
+	d, err := openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.Close()
+
+	cases := []struct {
+		name string
+		fsys fs.FS
+		want []string
+	}{
+		{"checksum changed", withEmbedded(t, fstest.MapFS{"0003_probe.sql": {Data: append([]byte("-- edited\n"), probeMigration.Data...)}}),
+			[]string{"0003", "0003_probe.sql", "checksum"}},
+		{"version missing from binary", migrationsFS(),
+			[]string{"0003", "0003_probe.sql", "missing from this binary"}},
+		{"name changed", withEmbedded(t, fstest.MapFS{"0003_renamed.sql": probeMigration}),
+			[]string{"0003", "0003_probe.sql", "0003_renamed.sql"}},
+	}
+	for _, c := range cases {
+		d, err := openWithMigrations(dsn, c.fsys)
+		if err == nil {
+			d.Close()
+			t.Errorf("%s: Open succeeded, want error", c.name)
+			continue
+		}
+		for _, w := range c.want {
+			if !strings.Contains(err.Error(), w) {
+				t.Errorf("%s: error %q does not mention %q", c.name, err, w)
+			}
+		}
+	}
+}
+
+func TestFailingMigrationRollsBackAndBlocksOpen(t *testing.T) {
+	dsn := scratchDSN(t, "fail")
+	fsys := withEmbedded(t, fstest.MapFS{
+		"0003_probe.sql": probeMigration,
+		"0004_broken.sql": {Data: []byte(`
+CREATE TABLE migration_broken (x INTEGER);
+SELECT 1/0;
+`)},
+	})
+	d, err := openWithMigrations(dsn, fsys)
+	if err == nil {
+		d.Close()
+		t.Fatal("Open succeeded with a failing migration")
+	}
+	if !strings.Contains(err.Error(), "0004") || !strings.Contains(err.Error(), "0004_broken.sql") {
+		t.Errorf("error %q does not name the failing migration", err)
+	}
+
+	// Inspect with the migrations that succeeded: earlier ones stay applied,
+	// the failing one left nothing behind.
+	d, err = openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	applied := readApplied(t, d)
+	if len(applied) != 3 || applied[2].version != 3 {
+		t.Fatalf("schema_migrations = %+v, want versions 1..3", applied)
+	}
+	var broken bool
+	if err := d.QueryRow(`SELECT to_regclass('migration_broken') IS NOT NULL`).Scan(&broken); err != nil {
+		t.Fatal(err)
+	}
+	if broken {
+		t.Error("migration_broken exists: the failed migration was not rolled back")
+	}
+}
+
+// A database already at 0003 must refuse a binary that adds 0002 below it,
+// before writing anything.
+func TestMigrationsRefuseOutOfOrderVersion(t *testing.T) {
+	dsn := scratchDSN(t, "outoforder")
+	records, err := fs.ReadFile(migrationsFS(), llmRecordsMigration)
+	if err != nil {
+		t.Fatal(err)
+	}
+	d, err := openWithMigrations(dsn, fstest.MapFS{
+		llmRecordsMigration: {Data: records},
+		"0003_probe.sql":    probeMigration,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	before := readApplied(t, d)
+	d.Close()
+	if len(before) != 2 || before[0].version != 1 || before[1].version != 3 {
+		t.Fatalf("setup schema_migrations = %+v, want versions 1 and 3", before)
+	}
+
+	d, err = openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err == nil {
+		d.Close()
+		t.Fatal("Open succeeded with 0002 pending below applied 0003")
+	}
+	for _, w := range []string{llmUsageMigration, "(0002)", "0003_probe.sql", "(0003)", "never inserted"} {
+		if !strings.Contains(err.Error(), w) {
+			t.Errorf("error %q does not mention %q", err, w)
+		}
+	}
+
+	raw, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer raw.Close()
+	after := readApplied(t, &DB{raw})
+	if len(after) != len(before) {
+		t.Fatalf("schema_migrations changed: before %+v, after %+v", before, after)
+	}
+	for i := range before {
+		if after[i].version != before[i].version || after[i].checksum != before[i].checksum || !after[i].appliedAt.Equal(before[i].appliedAt) {
+			t.Errorf("schema_migrations row %d changed: before %+v, after %+v", i, before[i], after[i])
+		}
+	}
+	var usage bool
+	if err := raw.QueryRow(`SELECT to_regclass('llm_usage') IS NOT NULL`).Scan(&usage); err != nil {
+		t.Fatal(err)
+	}
+	if usage {
+		t.Error("llm_usage exists: the refused 0002 migration ran")
+	}
+}
@@ -0,0 +1,29 @@
+
+CREATE TABLE IF NOT EXISTS llm_records (
+    id            BIGSERIAL PRIMARY KEY,
+    ts            TIMESTAMPTZ DEFAULT now(),
+    model         TEXT,
+    profile_name  TEXT,
+    session_id    TEXT,
+    task_id       TEXT,
+    worker        TEXT,
+    latency_ms    INTEGER,
+    input_tokens  INTEGER,
+    output_tokens INTEGER,
+    cache_read    INTEGER,
+    cache_write   INTEGER,
+    status        TEXT,
+    error         TEXT,
+    request_body  TEXT,
+    response_body TEXT,
+    raw_request   TEXT,
+    raw_response  TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_llm_records_ts ON llm_records(ts);
+CREATE INDEX IF NOT EXISTS idx_llm_records_session ON llm_records(session_id);
+
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS task_id TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS worker TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS profile_name TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_request TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_response TEXT;
@@ -0,0 +1,19 @@
+
+CREATE TABLE IF NOT EXISTS llm_usage (
+    id             BIGSERIAL PRIMARY KEY,
+    ts             TIMESTAMPTZ NOT NULL DEFAULT now(),
+    task_id        TEXT,
+    exploration_id BIGINT,
+    worker         TEXT,
+    model          TEXT,
+    profile_name   TEXT,
+    latency_ms     INTEGER,
+    input_tokens   INTEGER NOT NULL DEFAULT 0,
+    output_tokens  INTEGER NOT NULL DEFAULT 0,
+    cache_read     INTEGER NOT NULL DEFAULT 0,
+    cache_write    INTEGER NOT NULL DEFAULT 0,
+    status         TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_task  ON llm_usage(task_id);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_model ON llm_usage(task_id, model);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_exp   ON llm_usage(exploration_id);
@@ -447,6 +447,14 @@ func restoreArchiveAssets(tx *sql.Tx, taskID int64, raw json.RawMessage) (map[in
 			row["company_source"] = "explicit"
 			warnings = append(warnings, fmt.Sprintf("asset %d's company %d has been deleted; restored as unaffiliated", oldID, companyID))
 		}
+		// Archives taken before normalizeURL stripped userinfo can hold
+		// http://user:pass@host URLs. Normalise like the live upserts so the
+		// restore neither stores the credential nor misses the clean row.
+		if typeName, _ := row["type"].(string); typeName == "service" || typeName == "endpoint" {
+			if rawURL, ok := row["url"].(string); ok {
+				row["url"] = normalizeURL(rawURL)
+			}
+		}
 		if existing, found, err := findArchiveAssetNaturalID(tx, row); err != nil {
 			return nil, nil, err
 		} else if found {
@@ -0,0 +1,284 @@
+package db
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+)
+
+const restoreURLHost = "restore-userinfo.example"
+
+// bigintArray renders ids as a PostgreSQL array literal for a $n::bigint[] cast.
+func bigintArray(ids ...int64) string {
+	parts := make([]string, len(ids))
+	for i, id := range ids {
+		parts[i] = fmt.Sprint(id)
+	}
+	return "{" + strings.Join(parts, ",") + "}"
+}
+
+// archivedAssetRows exports live asset rows the way SnapshotTaskArchive does,
+// so the restore tests start from realistic archive JSON.
+func archivedAssetRows(t *testing.T, d *DB, ids ...int64) []map[string]any {
+	t.Helper()
+	raw, _, err := queryArchiveRows(d, `SELECT asset.* FROM assets asset WHERE asset.id = ANY($1::bigint[]) ORDER BY asset.id`, bigintArray(ids...))
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows, err := decodeArchiveRows(raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != len(ids) {
+		t.Fatalf("exported %d archive rows, want %d", len(rows), len(ids))
+	}
+	return rows
+}
+
+// restoreArchivedAssets runs restoreArchiveAssets on rows in a committed tx.
+func restoreArchivedAssets(t *testing.T, d *DB, taskID int64, rows []map[string]any) map[int64]int64 {
+	t.Helper()
+	raw, err := json.Marshal(rows)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tx, err := d.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback() //nolint:errcheck
+	mapping, _, err := restoreArchiveAssets(tx, taskID, raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	return mapping
+}
+
+func restoreURLCleanup(d *DB) {
+	d.Exec(`DELETE FROM assets WHERE url LIKE '%` + restoreURLHost + `%' OR domain LIKE '%` + restoreURLHost + `' OR root_domain = '` + restoreURLHost + `' OR ip = '198.51.100.231'`) //nolint:errcheck
+}
+
+func countRestoreURLRows(t *testing.T, d *DB, assetType, cleanURL string) int {
+	t.Helper()
+	var n int
+	if err := d.QueryRow(`SELECT count(*) FROM assets WHERE type=$1 AND (url=$2 OR url LIKE '%@%' || $3::text || '%')`, assetType, cleanURL, restoreURLHost).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+func assertNoRestoreURLCredential(t *testing.T, d *DB, id int64) {
+	t.Helper()
+	var row string
+	if err := d.QueryRow(`SELECT row_to_json(a)::text FROM assets a WHERE id=$1`, id).Scan(&row); err != nil {
+		t.Fatal(err)
+	}
+	for _, secret := range []string{"alice", "bob", "s3cr3t"} {
+		if strings.Contains(row, secret) {
+			t.Fatalf("restored asset %d still carries %q: %s", id, secret, row)
+		}
+	}
+}
+
+func TestRestoreArchiveAssetsStripsURLUserinfo(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	restoreURLCleanup(d)
+	defer restoreURLCleanup(d)
+	const taskID = int64(880000001)
+
+	svcURL := "http://" + restoreURLHost + ":8080/admin"
+	epURL := "https://api." + restoreURLHost + "/users?id=1"
+	svcID, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: svcURL, TaskID: taskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	epID, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: epURL, Method: "GET", TaskID: taskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows := archivedAssetRows(t, d, svcID, epID)
+	// Simulate an archive taken before normalizeURL dropped userinfo.
+	rows[0]["url"] = "http://alice:s3cr3t-one@" + restoreURLHost + ":8080/admin"
+	rows[1]["url"] = "https://bob:s3cr3t-two@api." + restoreURLHost + "/users?id=1"
+	archived := map[string]map[string]any{"service": {}, "endpoint": {}}
+	for _, row := range rows {
+		typeName := row["type"].(string)
+		for _, key := range []string{"domain", "root_domain", "port", "service_name", "method"} {
+			archived[typeName][key] = row[key]
+		}
+	}
+	if _, err := d.Exec(`DELETE FROM assets WHERE id = ANY($1::bigint[])`, bigintArray(svcID, epID)); err != nil {
+		t.Fatal(err)
+	}
+
+	mapping := restoreArchivedAssets(t, d, taskID, rows)
+	for oldID, wantURL := range map[int64]string{svcID: svcURL, epID: epURL} {
+		newID := mapping[oldID]
+		if newID == 0 {
+			t.Fatalf("asset %d was not restored: %v", oldID, mapping)
+		}
+		var gotURL, typeName string
+		var domain, rootDomain, serviceName, method *string
+		var port *int64
+		if err := d.QueryRow(`SELECT url,type,domain,root_domain,port,service_name,method FROM assets WHERE id=$1`, newID).
+			Scan(&gotURL, &typeName, &domain, &rootDomain, &port, &serviceName, &method); err != nil {
+			t.Fatal(err)
+		}
+		if gotURL != wantURL {
+			t.Fatalf("restored %s url=%q, want %q", typeName, gotURL, wantURL)
+		}
+		assertNoRestoreURLCredential(t, d, newID)
+		// Only url is normalised: the other columns keep the archived values.
+		want := archived[typeName]
+		got := map[string]any{"domain": domain, "root_domain": rootDomain, "port": port, "service_name": serviceName, "method": method}
+		for key, value := range got {
+			gotJSON, _ := json.Marshal(value)
+			wantJSON, _ := json.Marshal(want[key])
+			if string(gotJSON) != string(wantJSON) {
+				t.Errorf("%s %s=%s, want archived %s", typeName, key, gotJSON, wantJSON)
+			}
+		}
+		if n := countRestoreURLRows(t, d, typeName, wantURL); n != 1 {
+			t.Fatalf("%s rows for %q = %d, want 1", typeName, wantURL, n)
+		}
+	}
+}
+
+func TestRestoreArchiveAssetsReusesCleanURLRow(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	restoreURLCleanup(d)
+	defer restoreURLCleanup(d)
+	const liveTaskID, archivedTaskID = int64(880000002), int64(880000003)
+
+	svcURL := "http://" + restoreURLHost + ":8080/admin"
+	epURL := "https://api." + restoreURLHost + "/users?id=1"
+	svcID, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: svcURL, TaskID: liveTaskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	epID, err := av2.UpsertEndpoint(UpsertEndpointReq{URL: epURL, Method: "GET", TaskID: liveTaskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows := archivedAssetRows(t, d, svcID, epID)
+	// The archived rows are other, older rows whose url still carries userinfo.
+	rows[0]["id"], rows[0]["url"] = int64(990000101), "http://alice:s3cr3t-one@"+restoreURLHost+":8080/admin"
+	rows[1]["id"], rows[1]["url"] = int64(990000102), "https://bob:s3cr3t-two@api."+restoreURLHost+"/users?id=1"
+	for _, row := range rows {
+		row["task_ids"] = []any{archivedTaskID}
+	}
+
+	mapping := restoreArchivedAssets(t, d, archivedTaskID, rows)
+	for oldID, existingID := range map[int64]int64{990000101: svcID, 990000102: epID} {
+		if mapping[oldID] != existingID {
+			t.Fatalf("archived asset %d mapped to %d, want existing clean row %d", oldID, mapping[oldID], existingID)
+		}
+		var taskIDs string
+		if err := d.QueryRow(`SELECT task_ids::text FROM assets WHERE id=$1`, existingID).Scan(&taskIDs); err != nil {
+			t.Fatal(err)
+		}
+		if want := bigintArray(liveTaskID, archivedTaskID); taskIDs != want {
+			t.Fatalf("asset %d task_ids=%s, want %s", existingID, taskIDs, want)
+		}
+		assertNoRestoreURLCredential(t, d, existingID)
+		var leftover int
+		if err := d.QueryRow(`SELECT count(*) FROM assets WHERE id=$1`, oldID).Scan(&leftover); err != nil {
+			t.Fatal(err)
+		}
+		if leftover != 0 {
+			t.Fatalf("archived asset %d inserted beside the clean row", oldID)
+		}
+	}
+	if n := countRestoreURLRows(t, d, "service", svcURL); n != 1 {
+		t.Fatalf("service rows = %d, want 1", n)
+	}
+	if n := countRestoreURLRows(t, d, "endpoint", epURL); n != 1 {
+		t.Fatalf("endpoint rows = %d, want 1", n)
+	}
+}
+
+func TestRestoreArchiveAssetsDedupsURLAcrossCredentials(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	restoreURLCleanup(d)
+	defer restoreURLCleanup(d)
+	const taskID = int64(880000004)
+
+	svcURL := "http://" + restoreURLHost + ":8080/admin"
+	svcID, err := av2.UpsertHTTPService(UpsertHTTPServiceReq{URL: svcURL, TaskID: taskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	first := archivedAssetRows(t, d, svcID)[0]
+	second := archivedAssetRows(t, d, svcID)[0]
+	if _, err := d.Exec(`DELETE FROM assets WHERE id=$1`, svcID); err != nil {
+		t.Fatal(err)
+	}
+	first["id"], first["url"] = int64(990000201), "http://alice:s3cr3t-one@"+restoreURLHost+":8080/admin"
+	second["id"], second["url"] = int64(990000202), "http://bob:s3cr3t-two@"+restoreURLHost+":8080/admin"
+
+	mapping := restoreArchivedAssets(t, d, taskID, []map[string]any{first, second})
+	if mapping[990000201] == 0 || mapping[990000201] != mapping[990000202] {
+		t.Fatalf("credential variants restored as separate rows: %v", mapping)
+	}
+	if n := countRestoreURLRows(t, d, "service", svcURL); n != 1 {
+		t.Fatalf("service rows = %d, want exactly 1", n)
+	}
+	assertNoRestoreURLCredential(t, d, mapping[990000201])
+}
+
+func TestRestoreArchiveAssetsLeavesDomainAndIPURLUntouched(t *testing.T) {
+	d, av2, _ := testSetup(t)
+	defer d.Close()
+	restoreURLCleanup(d)
+	defer restoreURLCleanup(d)
+	const taskID = int64(880000005)
+
+	domainID, err := av2.UpsertRootDomain(UpsertRootDomainReq{Domain: restoreURLHost, TaskID: taskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ipID, err := av2.UpsertIP(UpsertIPReq{IP: "198.51.100.231", TaskID: taskID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rows := archivedAssetRows(t, d, domainID, ipID)
+	// A url normalizeURL would rewrite proves the normalisation is scoped to
+	// service and endpoint rows.
+	const oddURL = "HTTP://Restore-Userinfo.example/"
+	for _, row := range rows {
+		row["url"] = oddURL
+	}
+	want := map[int64][]any{}
+	for _, row := range rows {
+		id, _ := jsonInt64(row["id"])
+		want[id] = []any{row["type"], row["domain"], row["root_domain"], row["ip"]}
+	}
+	if _, err := d.Exec(`DELETE FROM assets WHERE id = ANY($1::bigint[])`, bigintArray(domainID, ipID)); err != nil {
+		t.Fatal(err)
+	}
+
+	mapping := restoreArchivedAssets(t, d, taskID, rows)
+	for oldID, fields := range want {
+		var typeName string
+		var domain, rootDomain, ip, gotURL *string
+		if err := d.QueryRow(`SELECT type,domain,root_domain,ip,url FROM assets WHERE id=$1`, mapping[oldID]).
+			Scan(&typeName, &domain, &rootDomain, &ip, &gotURL); err != nil {
+			t.Fatal(err)
+		}
+		got, _ := json.Marshal([]any{typeName, domain, rootDomain, ip})
+		wantJSON, _ := json.Marshal(fields)
+		if string(got) != string(wantJSON) {
+			t.Errorf("asset %d restored as %s, want %s", oldID, got, wantJSON)
+		}
+		if gotURL == nil || *gotURL != oddURL {
+			t.Errorf("%s asset url changed to %v, want %q", typeName, gotURL, oddURL)
+		}
+	}
+}
@@ -12,13 +12,14 @@ func TestRegisterTaskAssetScopesCreatesAssetsAndPersistsTextScope(t *testing.T)
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	task, err := d.CreateTask("manual scope registration", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	suffix := time.Now().UnixNano()
 	domain := fmt.Sprintf("manual-scope-%d.example.test", suffix)
 	ip := fmt.Sprintf("203.0.%d.%d", (suffix/250)%250, suffix%250+1)
@@ -48,7 +49,10 @@ func TestRegisterTaskAssetScopesCreatesAssetsAndPersistsTextScope(t *testing.T)
 			t.Fatalf("unexpected task asset provenance: %+v", asset)
 		}
 	}
-	t.Cleanup(func() { _, _ = d.Assets().DeleteByIDs(assetIDs) })
+	t.Cleanup(func() {
+		_, err := d.Assets().DeleteByIDs(assetIDs)
+		cleanupErr(t, "assets", err)
+	})
 	scopes, err := d.Assets().ListTaskScope(task.ID)
 	if err != nil || len(scopes) != 5 {
 		t.Fatalf("task scope=%+v err=%v", scopes, err)
@@ -91,18 +95,22 @@ func TestTaskAssetAttachDetachPreservesGlobalAssetAndAnchors(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	task, err := d.CreateTask("asset editing", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	assetID, err := d.Assets().UpsertRootDomain(UpsertRootDomainReq{Domain: fmt.Sprintf("manual-%d.example.test", time.Now().UnixNano())})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.Assets().DeleteByIDs([]int64{assetID}) })
+	t.Cleanup(func() {
+		_, err := d.Assets().DeleteByIDs([]int64{assetID})
+		cleanupErr(t, "asset", err)
+	})

 	mutation, err := d.Assets().AttachAssetsToTask(task.ID, []int64{assetID, assetID}, "authorized asset list item 3")
 	if err != nil {
@@ -144,7 +152,8 @@ func TestIntentAssetsIncludesDirectSourceProvenance(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	source, err := d.CreateTask("source assets", "goal", nil, 0, 0)
 	if err != nil {
@@ -154,7 +163,10 @@ func TestIntentAssetsIncludesDirectSourceProvenance(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(current.ID); _ = d.DeleteTask(source.ID) })
+	t.Cleanup(func() {
+		cleanupErr(t, "current task", d.DeleteTask(current.ID))
+		cleanupErr(t, "source task", d.DeleteTask(source.ID))
+	})

 	assetID, err := d.Assets().UpsertHTTPService(UpsertHTTPServiceReq{
 		URL: fmt.Sprintf("https://intent-%d.example.test", time.Now().UnixNano()), TaskID: source.ID,
@@ -162,7 +174,10 @@ func TestIntentAssetsIncludesDirectSourceProvenance(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.Assets().DeleteByIDs([]int64{assetID}) })
+	t.Cleanup(func() {
+		_, err := d.Assets().DeleteByIDs([]int64{assetID})
+		cleanupErr(t, "asset", err)
+	})
 	intentID, err := d.Exploration(source.ExplorationID).AddIntent(map[string]any{"summary": "source worker"}, 5, []int64{assetID}, "planner")
 	if err != nil {
 		t.Fatal(err)
@@ -13,7 +13,8 @@ func TestTaskCategoryCRUDAndTaskAssignment(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	name := fmt.Sprintf("Category %d", time.Now().UnixNano())
 	category, err := d.CreateTaskCategory("  " + strings.ReplaceAll(name, " ", "   ") + "  ")
@@ -31,7 +32,7 @@ func TestTaskCategoryCRUDAndTaskAssignment(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 	loaded, err := d.GetTask(task.ID)
 	if err != nil || loaded == nil || loaded.CategoryID == nil || *loaded.CategoryID != category.ID || loaded.CategoryName != name {
 		t.Fatalf("categorized task=%+v err=%v", loaded, err)
@@ -95,14 +96,18 @@ func TestSetTasksCategoryBatch(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	category, err := d.CreateTaskCategory(fmt.Sprintf("Batch %d", stamp))
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.DeleteTaskCategory(category.ID) })
+	t.Cleanup(func() {
+		_, err := d.DeleteTaskCategory(category.ID)
+		cleanupErr(t, "category", err)
+	})

 	ids := make([]int64, 0, 3)
 	for i := range 3 {
@@ -110,7 +115,7 @@ func TestSetTasksCategoryBatch(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+		t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })
 		ids = append(ids, task.ID)
 	}

@@ -152,14 +157,15 @@ func TestSetTasksCategoryRejectsBadInputAtomically(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	task, err := d.CreateTaskWithOptions(fmt.Sprintf("batch-reject-%d", stamp), "goal", TaskCreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })

 	missingCategory := int64(1 << 62)
 	if _, _, err := d.SetTasksCategory([]int64{task.ID}, &missingCategory); !errors.Is(err, ErrTaskCategoryNotFound) {
@@ -13,7 +13,8 @@ func TestTaskLLMProfileMutationsLockTaskBeforeProfile(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	first, err := d.SaveProfile(&LLMProfile{
@@ -38,9 +39,9 @@ func TestTaskLLMProfileMutationsLockTaskBeforeProfile(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(task.ID)
-		_ = d.DeleteProfile(first)
-		_ = d.DeleteProfile(second)
+		cleanupErr(t, "task", d.DeleteTask(task.ID))
+		cleanupProfile(t, d, first)
+		cleanupProfile(t, d, second)
 	})

 	replaceDB, replacePID := openSingleConnectionTestDB(t, dsn)
@@ -112,7 +113,8 @@ func TestDeleteProfileLocksNonTaskReferencesBeforeProfile(t *testing.T) {
 		if err != nil {
 			t.Skipf("postgres unavailable (%v) - skipping", err)
 		}
-		defer d.Close()
+		// Registered FIRST so it runs LAST, after the row cleanup below.
+		t.Cleanup(func() { d.Close() })

 		suffix := time.Now().UnixNano()
 		profileID, err := d.SaveProfile(&LLMProfile{
@@ -132,8 +134,8 @@ func TestDeleteProfileLocksNonTaskReferencesBeforeProfile(t *testing.T) {
 			t.Fatal(err)
 		}
 		t.Cleanup(func() {
-			_ = d.DeleteAgent(agent.Key)
-			_ = d.DeleteProfile(profileID)
+			cleanupErr(t, "agent", d.DeleteAgent(agent.Key))
+			cleanupProfile(t, d, profileID)
 		})

 		blocker, err := d.Begin()
@@ -169,7 +171,8 @@ func TestDeleteProfileLocksNonTaskReferencesBeforeProfile(t *testing.T) {
 		if err != nil {
 			t.Skipf("postgres unavailable (%v) - skipping", err)
 		}
-		defer d.Close()
+		// Registered FIRST so it runs LAST, after the row cleanup below.
+		t.Cleanup(func() { d.Close() })

 		suffix := time.Now().UnixNano()
 		profileID, err := d.SaveProfile(&LLMProfile{
@@ -184,8 +187,8 @@ func TestDeleteProfileLocksNonTaskReferencesBeforeProfile(t *testing.T) {
 			t.Fatal(err)
 		}
 		t.Cleanup(func() {
-			_ = d.DeleteConversation(conversation.ID)
-			_ = d.DeleteProfile(profileID)
+			cleanupErr(t, "conversation", d.DeleteConversation(conversation.ID))
+			cleanupProfile(t, d, profileID)
 		})

 		blocker, err := d.Begin()
@@ -225,7 +228,8 @@ func TestNonTaskProfileMutationsLockReferenceBeforeProfile(t *testing.T) {
 		if err != nil {
 			t.Skipf("postgres unavailable (%v) - skipping", err)
 		}
-		defer d.Close()
+		// Registered FIRST so it runs LAST, after the row cleanup below.
+		t.Cleanup(func() { d.Close() })

 		suffix := time.Now().UnixNano()
 		profileID, err := d.SaveProfile(&LLMProfile{
@@ -240,8 +244,8 @@ func TestNonTaskProfileMutationsLockReferenceBeforeProfile(t *testing.T) {
 			t.Fatal(err)
 		}
 		t.Cleanup(func() {
-			_ = d.DeleteAgent(agent.Key)
-			_ = d.DeleteProfile(profileID)
+			cleanupErr(t, "agent", d.DeleteAgent(agent.Key))
+			cleanupProfile(t, d, profileID)
 		})

 		blocker, err := d.Begin()
@@ -273,7 +277,8 @@ func TestNonTaskProfileMutationsLockReferenceBeforeProfile(t *testing.T) {
 		if err != nil {
 			t.Skipf("postgres unavailable (%v) - skipping", err)
 		}
-		defer d.Close()
+		// Registered FIRST so it runs LAST, after the row cleanup below.
+		t.Cleanup(func() { d.Close() })

 		suffix := time.Now().UnixNano()
 		profileID, err := d.SaveProfile(&LLMProfile{
@@ -288,8 +293,8 @@ func TestNonTaskProfileMutationsLockReferenceBeforeProfile(t *testing.T) {
 			t.Fatal(err)
 		}
 		t.Cleanup(func() {
-			_ = d.DeleteConversation(conversation.ID)
-			_ = d.DeleteProfile(profileID)
+			cleanupErr(t, "conversation", d.DeleteConversation(conversation.ID))
+			cleanupProfile(t, d, profileID)
 		})

 		blocker, err := d.Begin()
@@ -323,7 +328,8 @@ func TestCreateConversationAndDeleteProfileDoNotDeadlock(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	profileID, err := d.SaveProfile(&LLMProfile{
@@ -334,8 +340,8 @@ func TestCreateConversationAndDeleteProfileDoNotDeadlock(t *testing.T) {
 	}
 	title := fmt.Sprintf("conversation create race %d", suffix)
 	t.Cleanup(func() {
-		_, _ = d.Exec(`DELETE FROM conversations WHERE title=$1`, title)
-		_ = d.DeleteProfile(profileID)
+		cleanupExec(t, d, `DELETE FROM conversations WHERE title=$1`, title)
+		cleanupProfile(t, d, profileID)
 	})

 	profileBlocker, err := d.Begin()
@@ -403,7 +409,8 @@ func TestDeleteProfileRetriesReferenceCommittedAfterInitialScan(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	profileID, err := d.SaveProfile(&LLMProfile{
@@ -418,8 +425,8 @@ func TestDeleteProfileRetriesReferenceCommittedAfterInitialScan(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(task.ID)
-		_ = d.DeleteProfile(profileID)
+		cleanupErr(t, "task", d.DeleteTask(task.ID))
+		cleanupProfile(t, d, profileID)
 	})

 	// Keep the new reference uncommitted while deletion takes its initial
@@ -11,7 +11,8 @@ func TestListTasksBulkHydratesTaskContext(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	profileID, err := d.SaveProfile(&LLMProfile{
@@ -20,17 +21,17 @@ func TestListTasksBulkHydratesTaskContext(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteProfile(profileID) })
+	t.Cleanup(func() { cleanupProfile(t, d, profileID) })
 	source, err := d.CreateTask(fmt.Sprintf("bulk-list-source-%d", stamp), "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(source.ID) })
+	t.Cleanup(func() { cleanupErr(t, "source task", d.DeleteTask(source.ID)) })
 	companyID, _, err := d.Companies().UpsertCompany(fmt.Sprintf("Bulk List Company %d", stamp), "")
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.Companies().DeleteCompany(companyID) })
+	t.Cleanup(func() { cleanupErr(t, "company", d.Companies().DeleteCompany(companyID)) })
 	child, err := d.CreateTaskWithOptions("bulk-list-child", "goal", TaskCreateOptions{
 		LLMProfileIDs: []int64{profileID},
 		SourceTaskIDs: []int64{source.ID},
@@ -39,7 +40,7 @@ func TestListTasksBulkHydratesTaskContext(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(child.ID) })
+	t.Cleanup(func() { cleanupErr(t, "child task", d.DeleteTask(child.ID)) })

 	tasks, err := d.ListTasks()
 	if err != nil {
@@ -72,13 +73,14 @@ func TestTaskListMetricsAll(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	task, err := d.CreateTask("task-list-metrics", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "task", d.DeleteTask(task.ID)) })

 	resultAt := time.Now().Add(-time.Second).Truncate(time.Second)
 	latestAt := resultAt.Add(time.Second)
@@ -14,7 +14,8 @@ func TestTaskTemplateCRUDAndNormalizedUniqueness(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	name := fmt.Sprintf("Template %d", suffix)
@@ -26,7 +27,10 @@ func TestTaskTemplateCRUDAndNormalizedUniqueness(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.DeleteTaskTemplate(created.ID) })
+	t.Cleanup(func() {
+		_, err := d.DeleteTaskTemplate(created.ID)
+		cleanupErr(t, "template", err)
+	})
 	if created.Name != name || created.Description != "initial description" || created.Goal != "initial goal" {
 		t.Fatalf("template was not normalized: %+v", created)
 	}
@@ -85,7 +89,8 @@ func TestTaskTemplateDisjointPatchesCompose(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanup below.
+	t.Cleanup(func() { d.Close() })

 	created, err := d.CreateTaskTemplate(TaskTemplateInput{
 		Name:        fmt.Sprintf("Concurrent template %d", time.Now().UnixNano()),
@@ -95,7 +100,10 @@ func TestTaskTemplateDisjointPatchesCompose(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.DeleteTaskTemplate(created.ID) })
+	t.Cleanup(func() {
+		_, err := d.DeleteTaskTemplate(created.ID)
+		cleanupErr(t, "template", err)
+	})

 	description := "description from concurrent patch"
 	goal := "goal from concurrent patch"
@@ -13,7 +13,7 @@ func TestTaskLifecycleAndDeleteCascade(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	t.Cleanup(func() { d.Close() })

 	tk, err := d.CreateTask("migration test", "goal X", nil, 0, 0)
 	if err != nil {
@@ -87,7 +87,8 @@ func TestTaskDeleteCascadeAssets(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	first, err := d.CreateTask("cascade delete test", "goal A", nil, 0, 0)
 	if err != nil {
@@ -138,9 +139,10 @@ func TestTaskDeleteCascadeAssets(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(first.ID)
-		_ = d.DeleteTask(second.ID)
-		_, _ = assets.DeleteByIDs([]int64{exclusiveID, sharedID, anchorOnlyID, protectedID})
+		cleanupErr(t, "first task", d.DeleteTask(first.ID))
+		cleanupErr(t, "second task", d.DeleteTask(second.ID))
+		_, err := assets.DeleteByIDs([]int64{exclusiveID, sharedID, anchorOnlyID, protectedID})
+		cleanupErr(t, "assets", err)
 	})

 	hosts, err := assets.HostsByTask(first.ID)
@@ -211,7 +213,8 @@ func TestTaskRelationsAndLLMFailoverChain(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	profileIDs := make([]int64, 0, 3)
@@ -242,11 +245,11 @@ func TestTaskRelationsAndLLMFailoverChain(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_ = d.DeleteTask(child.ID)
-		_ = d.DeleteTask(sourceA.ID)
-		_ = d.DeleteTask(sourceB.ID)
+		cleanupErr(t, "child task", d.DeleteTask(child.ID))
+		cleanupErr(t, "source task A", d.DeleteTask(sourceA.ID))
+		cleanupErr(t, "source task B", d.DeleteTask(sourceB.ID))
 		for _, id := range profileIDs {
-			_ = d.DeleteProfile(id)
+			cleanupProfile(t, d, id)
 		}
 	})

@@ -469,13 +472,14 @@ func TestTaskContextRejectsDuplicatesAndAllowsTerminalLLMEdits(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	source, err := d.CreateTask("source", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(source.ID) })
+	t.Cleanup(func() { cleanupErr(t, "source task", d.DeleteTask(source.ID)) })
 	if _, err := d.CreateTaskWithOptions("bad", "goal", TaskCreateOptions{SourceTaskIDs: []int64{source.ID, source.ID}}); err == nil {
 		t.Fatal("duplicate source task ids should be rejected")
 	}
@@ -490,13 +494,13 @@ func TestTaskContextRejectsDuplicatesAndAllowsTerminalLLMEdits(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteProfile(profileID) })
+	t.Cleanup(func() { cleanupProfile(t, d, profileID) })

 	task, err := d.CreateTask("terminal", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "terminal task", d.DeleteTask(task.ID)) })
 	if err := d.SetStatus(task.ID, "done"); err != nil {
 		t.Fatal(err)
 	}
@@ -523,7 +527,8 @@ func TestCreateTaskWithCompanyScopes(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) - skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	suffix := time.Now().UnixNano()
 	companyA, _, err := d.Companies().UpsertCompany(fmt.Sprintf("Task Company A %d", suffix), "")
@@ -539,14 +544,14 @@ func TestCreateTaskWithCompanyScopes(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		_, _ = d.Exec(`DELETE FROM companies WHERE id IN ($1,$2,$3)`, companyA, companyB, emptyCompany)
+		cleanupExec(t, d, `DELETE FROM companies WHERE id IN ($1,$2,$3)`, companyA, companyB, emptyCompany)
 	})

 	existingTask, err := d.CreateTask("existing company asset owner", "goal", nil, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(existingTask.ID) })
+	t.Cleanup(func() { cleanupErr(t, "existing task", d.DeleteTask(existingTask.ID)) })
 	var companyAssetA, companyAssetB int64
 	domainA := fmt.Sprintf("task-company-a-%d.example.test", suffix)
 	domainB := fmt.Sprintf("task-company-b-%d.example.test", suffix)
@@ -562,7 +567,10 @@ VALUES ('root_domain',$1,$1,$2,'explicit')
 RETURNING id`, domainB, companyB).Scan(&companyAssetB); err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _, _ = d.Assets().DeleteByIDs([]int64{companyAssetA, companyAssetB}) })
+	t.Cleanup(func() {
+		_, err := d.Assets().DeleteByIDs([]int64{companyAssetA, companyAssetB})
+		cleanupErr(t, "company assets", err)
+	})

 	task, err := d.CreateTaskWithOptions("company-scoped task", "use company scope", TaskCreateOptions{
 		CompanyIDs: []int64{companyA, companyB},
@@ -570,7 +578,7 @@ RETURNING id`, domainB, companyB).Scan(&companyAssetB); err != nil {
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
+	t.Cleanup(func() { cleanupErr(t, "company-scoped task", d.DeleteTask(task.ID)) })
 	if fmt.Sprint(task.CompanyIDs) != fmt.Sprint([]int64{companyA, companyB}) {
 		t.Fatalf("creation result company IDs=%v", task.CompanyIDs)
 	}
@@ -674,7 +682,8 @@ func TestListTasksOrderByIDDesc(t *testing.T) {
 	if err != nil {
 		t.Skipf("postgres unavailable (%v) — skipping", err)
 	}
-	defer d.Close()
+	// Registered FIRST so it runs LAST, after the row cleanups below.
+	t.Cleanup(func() { d.Close() })

 	stamp := time.Now().UnixNano()
 	var ids []int64
@@ -683,7 +692,7 @@ func TestListTasksOrderByIDDesc(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		t.Cleanup(func() { _ = d.DeleteTask(tk.ID) })
+		t.Cleanup(func() { cleanupErr(t, "order task", d.DeleteTask(tk.ID)) })
 		ids = append(ids, tk.ID)
 	}
Author	SHA1	Message	Date
BaaRai	f3b9f6f688	db: refuse a pending migration older than the newest applied one applyMigrations now fails closed, before any write, when a pending migration has a version below the highest version recorded in schema_migrations: migrations must be appended, never inserted.	2026-11-14 13:10:57 +01:00
BaaRai	1da0d07222	db: versioned migrations runner with schema_migrations Move the llm_records and llm_usage DDL out of Go constants into embedded files db/migrations/0001_llm_records.sql and 0002_llm_usage.sql (byte-for-byte the old constants). Open now applies pending migrations after schema.sql and before seedBuiltins, under the existing schema advisory lock and on the lock-holding connection. Each migration runs in its own transaction with its schema_migrations row (version, name, SHA-256 checksum, applied_at). Open fails closed when the database has a version missing from the binary, or an applied migration whose file name or checksum changed. A failing migration is rolled back and blocks startup; earlier ones stay applied. EnsureLLMRecordsTable and EnsureLLMUsageTable re-run the embedded files and keep their transaction-level schema lock; Open never calls them.	2026-11-14 12:52:34 +01:00
BaaRai	8f3f0c54f8	Merge pull request #224 from BaaRai/fix/archive-restore-normalize-url db: normalise archived asset URLs on restore	2026-11-14 12:27:39 +01:00
BaaRai	2e43b423ee	db: normalise archived asset URLs on restore restoreArchiveAssets inserted archived service and endpoint rows with their raw url, and findArchiveAssetNaturalID matched on that raw value. An archive taken before normalizeURL dropped userinfo therefore restored http://user:pass@host in clear and duplicated the existing clean row. Pass row["url"] through normalizeURL for service and endpoint rows before the natural-identity lookup, so the lookup reuses the clean row and an insert stores the normalised URL. Other columns keep their archived values.	2026-11-14 12:17:33 +01:00
BaaRai	8bb4ca0745	Merge pull request #223 from BaaRai/fix/store-normalize-userinfo db: drop URL userinfo in normalizeURL before storing assets.url	2026-11-14 12:06:44 +01:00
BaaRai	939f8d4976	db: drop URL userinfo in normalizeURL before storing assets.url UpsertHTTPService and UpsertEndpoint both key assets.url on normalizeURL, which lowercased scheme and host but kept u.User. A URL passed with user:pass@ (insert_assets, server imports, sync) was stored verbatim and the same URL with different credentials produced separate rows. Set u.User = nil before serialising. Scheme, host, port, path and query are unchanged; an @ in the path or query is preserved. Tests: normalizeURL unit cases, plus PG tests for both upserts covering credential stripping, dedup across different credentials, @ in path, and an unchanged URL without userinfo.	2026-11-14 11:56:43 +01:00
BaaRai	b5c0bb9f33	Merge pull request #222 from BaaRai/fix/scan-ingest-strip-userinfo fix(agent): strip URL userinfo in scan and bbot ingestion	2026-11-14 01:12:17 +01:00
BaaRai	9d75df335f	fix(agent): strip URL userinfo in scan and bbot ingestion scanURL kept the userinfo of httpx, ffuf and nuclei URLs, so a line such as http://user:pass@host/path stored the credentials in assets.url, in the notable items and in the facts. scanURL now drops u.User before re-serializing; scheme, host, port, path and query are kept. The nuclei fact detail also embedded the raw host and matched-at values, the httpx detail the raw host and the ffuf detail the raw redirect location. They now go through stripUserinfo, which removes only the userinfo of a URL or bare host:port and keeps every other byte, so a value without userinfo is unchanged and fact deduplication is the same. bbotTyped drops the userinfo of the typed url of an in-scope URL event before it becomes an asset. Rows already stored with userinfo are not rewritten here.	2026-11-14 01:02:16 +01:00
BaaRai	fe9814c5f6	Merge pull request #221 from BaaRai/chore/db-test-cleanup-order test(db): close the pool after the row cleanups, report cleanup errors	2026-11-14 00:51:55 +01:00
BaaRai	53fc8478da	test(db): close the pool after the row cleanups, report cleanup errors A `defer d.Close()` runs when the test function returns, before its t.Cleanup functions. The cleanups that delete the test's rows then ran on a closed pool, the error was ignored, and the rows stayed in the database for the next run. PR #220 fixed three files; this applies the same pattern to every remaining db test that combines `defer d.Close()` with row cleanups. - Close the pool with t.Cleanup, registered before the row cleanups so it runs last (LIFO). - Report cleanup errors with t.Errorf through small helpers in db/cleanup_helpers_test.go. A profile the test already deleted is not an error. Test files only. No assertion or production change.	2026-11-14 00:38:00 +01:00
BaaRai	edc2f4da6d	Merge pull request #220 from BaaRai/fix/db-test-isolation test(db): make profile and company-scope tests repeatable on one database	2026-11-14 00:19:41 +01:00