playbook_lift: harness expansion + reality test #001 (7/8 lift, 87.5%)

The 5-loop substrate's load-bearing gate is verified — playbook +
matrix indexer give the results we're looking for. Per the report's
rubric, lift ≥ 50% of discoveries means matrix is doing real work;
7/8 = 87.5% blew through that.

Harness was structurally hiding bugs behind a 5-daemon stripped boot.
Expanding to the full 10-daemon prod stack surfaced 7 fixes in cascade:

1. driver→matrixd: {"query": ...} → {"query_text": ...} field name
2. harness temp toml missing [s3] → wrong default bucket → catalogd
   rehydrate 500 on first call
3. harness→queryd SQL probe: {"q": ...} → {"sql": ...} field name
4. expand boot from 5 → 10 daemons in dep-ordered launch
5. add SQL surface probe (3-row CSV ingest → COUNT(*)=3 assertion)
6. candidates corpus was synthetic SWE-tech (Swift/iOS, Scala/Spark) —
   wrong domain for staffing queries; replaced with ethereal_workers
   (10K rows, real staffing schema, "e-" id prefix to avoid collision
   with workers' "w-"). staffing_workers driver gains -index-name +
   -id-prefix flags so the same binary serves both corpora
7. local_judge qwen3.5:latest is a vision-SSM 256K-ctx build running
   ~30s per judge call against the lift loop; reverted to
   qwen2.5:latest (~1s/call, 30× faster, held lift theory)

Each contract drift (1, 3) is now locked into a cmd/<bin>/main_test.go
so future drift fires in `go test`, not in a reality run. R-005 closed:

- cmd/matrixd/main_test.go (new) — playbook record drift detector +
  score bounds + 6 routes mounted
- cmd/queryd/main_test.go — wrong-field-name drift detector
- cmd/pathwayd/main_test.go (new) — 9 routes + add round-trip + retire
- cmd/observerd/main_test.go (new) — 4 routes + invalid-op + unknown-mode

`go test ./cmd/{matrixd,queryd,pathwayd,observerd}` all green.

Reality test results (reports/reality-tests/playbook_lift_001.{json,md}):
  Queries              21 (staffing-domain, 7 categories)
  Discoveries          8 (judge ≠ cosine top-1)
  Lifts                7/8 (87.5%)
  Boosts triggered     9
  Mean Δ distance      -0.053 (warm closer than cold)
  OOD honesty          dental/RN/SWE rated 1, no fake matches
  Cross-corpus boosts  confirmed (e- ↔ w- swaps in lifts)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scripts/staffing_workers/main.go

@@ -39,8 +39,7 @@ import (
 )
 
 const (
-	indexName = "workers"
-	dim       = 768
+	dim = 768
 )
 
 // workersSource implements corpusingest.Source over an in-memory
@@ -52,8 +51,9 @@ type workersSource struct {
 		workerID                                                            *chunkedInt64
 		name, role, city, state, skills, certs, archetype, resume, comm     *chunkedString
 	}
-	n   int64
-	cur int64
+	n        int64
+	cur      int64
+	idPrefix string // "w-" for workers, "e-" for ethereal_workers, etc.
 }
 
 // chunkedString lets per-row access work whether the table came back
@@ -120,7 +120,7 @@ func (c *chunkedInt64) At(row int64) int64 {
 	return 0
 }
 
-func newWorkersSource(path string) (*workersSource, func(), error) {
+func newWorkersSource(path, idPrefix string) (*workersSource, func(), error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return nil, nil, fmt.Errorf("open parquet: %w", err)
@@ -143,7 +143,7 @@ func newWorkersSource(path string) (*workersSource, func(), error) {
 		return nil, nil, fmt.Errorf("read table: %w", err)
 	}
 
-	src := &workersSource{n: table.NumRows()}
+	src := &workersSource{n: table.NumRows(), idPrefix: idPrefix}
 	schema := table.Schema()
 
 	stringCol := func(name string) (*chunkedString, error) {
@@ -248,7 +248,7 @@ func (s *workersSource) Next() (corpusingest.Row, error) {
 	text := b.String()
 
 	return corpusingest.Row{
-		ID:   fmt.Sprintf("w-%d", workerID),
+		ID:   fmt.Sprintf("%s%d", s.idPrefix, workerID),
 		Text: text,
 		Metadata: map[string]any{
 			"worker_id":      workerID,
@@ -267,15 +267,17 @@ func main() {
 	var (
 		gateway     = flag.String("gateway", "http://127.0.0.1:3110", "gateway base URL")
 		parquetPath = flag.String("parquet", "/home/profit/lakehouse/data/datasets/workers_500k.parquet", "workers parquet")
-		limit       = flag.Int("limit", 5000, "limit rows (0 = all 500K — usually not what you want here)")
-		drop        = flag.Bool("drop", true, "DELETE workers index before populate")
+		indexName   = flag.String("index-name", "workers", "vector index name (e.g. workers, ethereal_workers)")
+		idPrefix    = flag.String("id-prefix", "w-", "ID prefix to disambiguate worker_id collisions across corpora (e.g. w-, e-)")
+		limit       = flag.Int("limit", 5000, "limit rows (0 = all rows; default suits multi-corpus reality testing, not stress)")
+		drop        = flag.Bool("drop", true, "DELETE the index before populate")
 	)
 	flag.Parse()
 
 	hc := &http.Client{Timeout: 5 * time.Minute}
 	ctx := context.Background()
 
-	src, cleanup, err := newWorkersSource(*parquetPath)
+	src, cleanup, err := newWorkersSource(*parquetPath, *idPrefix)
 	if err != nil {
 		log.Fatalf("open workers source: %v", err)
 	}
@@ -283,7 +285,7 @@ func main() {
 
 	stats, err := corpusingest.Run(ctx, corpusingest.Config{
 		GatewayURL:   *gateway,
-		IndexName:    indexName,
+		IndexName:    *indexName,
 		Dimension:    dim,
 		Distance:     "cosine",
 		EmbedBatch:   16,
@@ -296,13 +298,13 @@ func main() {
 	}, src)
 	if err != nil {
 		if errors.Is(err, corpusingest.ErrPartialFailure) {
-			fmt.Printf("[workers] WARN partial failure: %v\n", err)
+			fmt.Printf("[%s] WARN partial failure: %v\n", *indexName, err)
 		} else {
 			log.Fatalf("ingest: %v", err)
 		}
 	}
-	fmt.Printf("[workers] populate: scanned=%d embedded=%d added=%d failed=%d wall=%v\n",
-		stats.Scanned, stats.Embedded, stats.Added, stats.FailedBatches,
+	fmt.Printf("[%s] populate: scanned=%d embedded=%d added=%d failed=%d wall=%v\n",
+		*indexName, stats.Scanned, stats.Embedded, stats.Added, stats.FailedBatches,
 		stats.Wall.Round(time.Millisecond))
 }