Batched Incorporation Implementation Plan

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

Goal: Let a collaborator select N≥2 open Topics on a Source and have the Agent incorporate them in one rewrite, one proposal, one commit — by modelling a batch as a Topic with anchor.kind = "batch" and a parent_topic_id link from each child.

Architecture: Add topics.parent_topic_id (nullable self-FK) and a new "batch" anchor kind. Drop the composite FK topics.(incorporated_proposal_id, id) → incorporation_proposals(id, topic_id) and enforce the looser invariant ("a Topic's incorporated proposal points at this Topic OR its parent") via a SQLite trigger + the existing collab write funnel. Every other piece — wb-agent, wb-incorporate, proposal freshness, the Approve handler, the realtime event surface — extends to handle the batch case where N=1 is the degenerate of N≥2.

Tech Stack: Go (server + wb-agent CLI + tests), SQLite (modernc.org/sqlite), plain JS/CSS (chrome.js), Claude Code (the Agent runtime), chromedp (e2e tests).

Spec: docs/superpowers/specs/2026-05-20-batched-incorporation-design.html

File map

Created

• internal/collab/migrations/006_batched_incorporation.sql — new column, dropped composite FK, triggers, index.
• internal/collab/batch.go — store helpers: CreateBatch, BatchParentForTopic, BatchChildrenOfParent, IsTopicBatched, DiscardBatch.
• internal/collab/batch_test.go — unit tests for the above.
• internal/server/handler_batches.go — POST /api/batches.
• internal/server/handler_batches_test.go — handler tests.

Modified

• internal/collab/anchor.go — add BatchAnchor{} type and "batch" kind.
• internal/collab/anchor_test.go — extend tests.
• internal/collab/mutators.go — CompleteIncorporation accepts N child topic IDs; ClaimIncorporation excludes batched children + batch parents.
• internal/collab/mutators_test.go — extend tests.
• internal/collab/freshness.go — openNonGlobalTopicIDsExcluding also excludes batched + batch parents.
• internal/collab/freshness_test.go — extend tests.
• internal/collab/incorporate.go — IncorporateInput passes child set; commit message gets one Topic: trailer per transitioned Topic.
• internal/collab/incorporate_test.go — extend tests.
• internal/collab/gitops.go — CommitInput.TopicIDs (slice) replaces TopicID (string) so trailers loop.
• internal/collab/gitops_test.go — extend tests.
• internal/collab/reader.go — TopicSummary carries ParentTopicID so the sidebar can render children under their parent.
• cmd/wb-agent/get_topic.go — when target Topic has anchor.kind = "batch", include children: [{topic, anchor, messages}].
• cmd/wb-agent/list_open_topics.go — replace --exclude-topic (singular) with --exclude-topics (CSV).
• cmd/wb-agent/main_test.go — extend tests.
• internal/server/handler_proposals.go — handleProposeRewrite 409s on a batched child; handleDiscardTopic clears children and writes audit reason when target is a batch parent; handleIncorporate transitions N+1 atomically + emits N+1 events.
• internal/server/handler_proposals_test.go — extend tests.
• internal/server/server.go — register POST /api/batches.
• internal/server/static/chrome.js — sidebar grouping with Batches section, multi-select mode, Resolve toolbar expander.
• internal/server/static/chrome.css — accent rail for children, multi-select header, batch pill.
• internal/server/static/chrome.css lives under internal/server/templates/ if the existing layout puts CSS there — check the actual layout when editing.
• internal/server/e2e_resolve_test.go — extend with a batch end-to-end test.
• .claude/skills/wb-incorporate/SKILL.md — branch on anchor.kind = "batch"; --exclude-topics plural; explanation contract covers N rewrites.

Conventions used by every task

• All Go tests live next to the file they test, in the same package (or _test sibling package where the existing file uses it).
• Run a single test with go test ./internal/collab/ -run TestBatchAnchor -v. Run the whole package with go test ./internal/collab/ -v. Run the full test suite with go test ./....
• The collab DB lives behind s.send(func(db *sql.DB) error { ... }) for writes — never bypass it from inside the same process. Reads can use s.db (or s.RawDB() from outside the package) directly.
• The Go module is github.com/getorcha/wiki-browser; the local working directory is /home/volrath/code/orcha/wiki-browser; the orcha repo root is one level above (cfg.Root).
• Each task ends with a commit. Commits are GPG-signed (default), no Co-Authored-By lines. Use a single-line git commit -m "..." per the user's convention.

Task 1: Migration 006 — add parent_topic_id, drop composite FK, add triggers

Files:

• Create: internal/collab/migrations/006_batched_incorporation.sql
• Test: internal/collab/migrate_test.go (extend) and internal/collab/batch_test.go (new)

The composite FK topics.(incorporated_proposal_id, id) → incorporation_proposals(id, topic_id) cannot be dropped via ALTER TABLE in SQLite. Use the twelve-step rebuild pattern from migration 003. Mirror the structure of migrations/003_agent_runtime.sql.

• Step 1: Write a failing migration test

Append the test to internal/collab/migrate_test.go (study the existing tests in that file for the pattern — they use collab.Open(...) and then s.RawDB().QueryRow(...) to inspect schema).

func TestMigration006_addsParentTopicAndDropsCompositeFK(t *testing.T) {
	dir := t.TempDir()
	s, err := Open(Config{Path: filepath.Join(dir, "collab.db")})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()

	// parent_topic_id column exists, nullable, FK to topics(id).
	rows, err := s.RawDB().Query(`PRAGMA table_info(topics)`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	var found bool
	for rows.Next() {
		var cid int
		var name, typ string
		var notnull int
		var dflt sql.NullString
		var pk int
		if err := rows.Scan(&cid, &name, &typ, &notnull, &dflt, &pk); err != nil {
			t.Fatal(err)
		}
		if name == "parent_topic_id" {
			found = true
			if notnull != 0 {
				t.Fatalf("parent_topic_id should be nullable, got notnull=%d", notnull)
			}
		}
	}
	if !found {
		t.Fatal("parent_topic_id column missing")
	}

	// Composite FK on (incorporated_proposal_id, id) is GONE.
	var sqlText string
	if err := s.RawDB().QueryRow(
		`SELECT sql FROM sqlite_master WHERE type='table' AND name='topics'`,
	).Scan(&sqlText); err != nil {
		t.Fatal(err)
	}
	if strings.Contains(sqlText, "REFERENCES incorporation_proposals(id, topic_id)") {
		t.Fatalf("composite FK still present in topics schema:\n%s", sqlText)
	}

	// Triggers exist. The "no grandparent" invariant is split across two
	// trigger names — one for UPDATE OF anchor, one for UPDATE OF parent_topic_id
	// (and a third pair on INSERT, see migration SQL below).
	wanted := []string{
		"topics_incorporated_proposal_match",
		"topics_parent_must_be_batch",
		"topics_parent_must_be_batch_insert",
		"topics_batch_no_grandparent_anchor",
		"topics_batch_no_grandparent_parent",
		"topics_batch_no_grandparent_insert",
	}
	for _, name := range wanted {
		var n int
		if err := s.RawDB().QueryRow(
			`SELECT COUNT(*) FROM sqlite_master WHERE type='trigger' AND name=?`, name,
		).Scan(&n); err != nil {
			t.Fatal(err)
		}
		if n != 1 {
			t.Fatalf("trigger %s missing", name)
		}
	}

	// Index exists.
	var idxCount int
	if err := s.RawDB().QueryRow(
		`SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='topics_parent_topic_id'`,
	).Scan(&idxCount); err != nil {
		t.Fatal(err)
	}
	if idxCount != 1 {
		t.Fatal("topics_parent_topic_id index missing")
	}
}

• Step 2: Run test to verify it fails

go test ./internal/collab/ -run TestMigration006 -v

Expected: FAIL (parent_topic_id column missing).

• Step 3: Write 006_batched_incorporation.sql

-- migrate:no-tx
-- Add topics.parent_topic_id, drop the composite FK
-- topics.(incorporated_proposal_id, id) → incorporation_proposals(id, topic_id),
-- and install triggers that enforce the new integrity invariants.
--
-- Idempotency: CREATE TRIGGER / CREATE INDEX use IF NOT EXISTS. The 12-step
-- rebuild itself is NOT idempotent across re-runs — the SELECT below hardcodes
-- parent_topic_id = NULL because the source `topics` table on first run has
-- no such column. Re-running this migration after any batch has been created
-- would wipe every parent_topic_id back to NULL. The migration runner's
-- schema_migrations bookkeeping is the only mechanism preventing re-runs;
-- never replay this file manually once it has applied successfully.

PRAGMA foreign_keys = OFF;
PRAGMA legacy_alter_table = OFF;

BEGIN;

CREATE TABLE topics_new (
  id                       TEXT PRIMARY KEY,
  source_path              TEXT NOT NULL,
  anchor                   TEXT,
  parent_topic_id          TEXT REFERENCES topics(id) ON DELETE RESTRICT,
  commit_sha               TEXT,
  incorporated_proposal_id TEXT,
  incorporated_by          TEXT,
  incorporated_at          INTEGER,
  discarded_by             TEXT,
  discarded_at             INTEGER,
  created_at               INTEGER NOT NULL,
  created_by               TEXT NOT NULL,
  updated_at               INTEGER NOT NULL,
  CHECK ((commit_sha               IS NULL) = (incorporated_at IS NULL)),
  CHECK ((incorporated_proposal_id IS NULL) = (incorporated_at IS NULL)),
  CHECK ((incorporated_by          IS NULL) = (incorporated_at IS NULL)),
  CHECK ((discarded_by             IS NULL) = (discarded_at    IS NULL)),
  CHECK (incorporated_at IS NULL OR discarded_at IS NULL),
  FOREIGN KEY (created_by)             REFERENCES users(id),
  FOREIGN KEY (incorporated_by)        REFERENCES users(id),
  FOREIGN KEY (discarded_by)           REFERENCES users(id),
  FOREIGN KEY (incorporated_proposal_id) REFERENCES incorporation_proposals(id)
);

INSERT INTO topics_new (
  id, source_path, anchor, parent_topic_id, commit_sha,
  incorporated_proposal_id, incorporated_by, incorporated_at,
  discarded_by, discarded_at, created_at, created_by, updated_at
)
SELECT id, source_path, anchor, NULL, commit_sha,
       incorporated_proposal_id, incorporated_by, incorporated_at,
       discarded_by, discarded_at, created_at, created_by, updated_at
  FROM topics;

DROP TABLE topics;
ALTER TABLE topics_new RENAME TO topics;

CREATE INDEX IF NOT EXISTS topics_by_source_path     ON topics(source_path);
CREATE INDEX IF NOT EXISTS topics_parent_topic_id    ON topics(parent_topic_id) WHERE parent_topic_id IS NOT NULL;

-- Replacement integrity trigger for the dropped composite FK. When a Topic
-- transitions to incorporated, the proposal it points at must be anchored on
-- the Topic itself OR on the Topic's parent (for a batched child).
CREATE TRIGGER IF NOT EXISTS topics_incorporated_proposal_match
AFTER UPDATE OF incorporated_proposal_id ON topics
WHEN NEW.incorporated_proposal_id IS NOT NULL
BEGIN
  SELECT CASE WHEN NOT EXISTS (
    SELECT 1 FROM incorporation_proposals p
    WHERE p.id = NEW.incorporated_proposal_id
      AND (p.topic_id = NEW.id OR p.topic_id = NEW.parent_topic_id)
  ) THEN RAISE(ABORT, 'incorporated_proposal_id does not belong to topic or its parent') END;
END;

-- A child's parent_topic_id must reference a Topic with anchor.kind = "batch".
-- Fires on both UPDATE and INSERT so a row that arrives with parent_topic_id
-- already populated also gets the check.
CREATE TRIGGER IF NOT EXISTS topics_parent_must_be_batch
AFTER UPDATE OF parent_topic_id ON topics
WHEN NEW.parent_topic_id IS NOT NULL
BEGIN
  SELECT CASE WHEN NOT EXISTS (
    SELECT 1 FROM topics p
    WHERE p.id = NEW.parent_topic_id
      AND json_extract(p.anchor, '$.kind') = 'batch'
  ) THEN RAISE(ABORT, 'parent_topic_id must reference a topic with anchor kind = batch') END;
END;

CREATE TRIGGER IF NOT EXISTS topics_parent_must_be_batch_insert
AFTER INSERT ON topics
WHEN NEW.parent_topic_id IS NOT NULL
BEGIN
  SELECT CASE WHEN NOT EXISTS (
    SELECT 1 FROM topics p
    WHERE p.id = NEW.parent_topic_id
      AND json_extract(p.anchor, '$.kind') = 'batch'
  ) THEN RAISE(ABORT, 'parent_topic_id must reference a topic with anchor kind = batch') END;
END;

-- A batch parent itself cannot have a parent (no nested batches). Guard on
-- UPDATE OF anchor, UPDATE OF parent_topic_id, and INSERT — covers every
-- mutation that could establish the forbidden combination.
CREATE TRIGGER IF NOT EXISTS topics_batch_no_grandparent_anchor
AFTER UPDATE OF anchor ON topics
WHEN json_extract(NEW.anchor, '$.kind') = 'batch' AND NEW.parent_topic_id IS NOT NULL
BEGIN
  SELECT RAISE(ABORT, 'a batch parent cannot itself be a child');
END;

CREATE TRIGGER IF NOT EXISTS topics_batch_no_grandparent_parent
AFTER UPDATE OF parent_topic_id ON topics
WHEN NEW.parent_topic_id IS NOT NULL AND json_extract(NEW.anchor, '$.kind') = 'batch'
BEGIN
  SELECT RAISE(ABORT, 'a batch parent cannot itself be a child');
END;

CREATE TRIGGER IF NOT EXISTS topics_batch_no_grandparent_insert
AFTER INSERT ON topics
WHEN NEW.parent_topic_id IS NOT NULL AND json_extract(NEW.anchor, '$.kind') = 'batch'
BEGIN
  SELECT RAISE(ABORT, 'a batch parent cannot itself be a child');
END;

PRAGMA foreign_key_check;

COMMIT;

PRAGMA foreign_keys = ON;

• Step 4: Run test to verify it passes

go test ./internal/collab/ -run TestMigration006 -v

Expected: PASS.

• Step 5: Add negative trigger tests

The triggers are the integrity contract that replaces the dropped composite FK. Confirming they fire (not just exist) is more valuable than the existence check from Step 1. Append to internal/collab/migrate_test.go:

func TestTrigger_topicsIncorporatedProposalMatch_fires(t *testing.T) {
	s, _ := newTestStore(t)
	if err := s.UpsertUser(User{ID: "u", DisplayName: "U"}); err != nil {
		t.Fatal(err)
	}
	// Two unrelated topics on different sources; the proposal belongs to t1,
	// but we try to point t2.incorporated_proposal_id at it.
	anchor, _ := MarshalAnchor(GlobalAnchor{})
	t1 := uuid.NewString()
	t2 := uuid.NewString()
	for _, id := range []string{t1, t2} {
		if err := s.InsertTopic(NewTopic{
			ID: id, SourcePath: "doc.md", Anchor: anchor, CreatedBy: "u",
		}); err != nil {
			t.Fatal(err)
		}
	}
	propID := uuid.NewString()
	if _, err := s.InsertProposal(NewProposal{
		ID: propID, TopicID: t1, RevisionNumber: 1,
		ProposedSource: "x", BaseSourceSHA: strings.Repeat("0", 40),
	}); err != nil {
		t.Fatal(err)
	}
	// Direct UPDATE bypassing CompleteIncorporation — the trigger must refuse.
	_, err := s.RawDB().Exec(
		`UPDATE topics
		    SET incorporated_proposal_id = ?, commit_sha = 'x',
		        incorporated_by = 'u', incorporated_at = 1
		  WHERE id = ?`,
		propID, t2,
	)
	if err == nil {
		t.Fatal("expected trigger to refuse mismatched proposal/topic pairing")
	}
	if !strings.Contains(err.Error(), "does not belong to topic or its parent") {
		t.Errorf("unexpected trigger message: %v", err)
	}
}

func TestTrigger_topicsParentMustBeBatch_fires(t *testing.T) {
	s, _ := newTestStore(t)
	if err := s.UpsertUser(User{ID: "u", DisplayName: "U"}); err != nil {
		t.Fatal(err)
	}
	// Two topics. Attempt to make t1 a child of t2 — but t2 has a global
	// anchor, not a batch anchor. The trigger must refuse.
	anchor, _ := MarshalAnchor(GlobalAnchor{})
	t1, t2 := uuid.NewString(), uuid.NewString()
	for _, id := range []string{t1, t2} {
		if err := s.InsertTopic(NewTopic{
			ID: id, SourcePath: "doc.md", Anchor: anchor, CreatedBy: "u",
		}); err != nil {
			t.Fatal(err)
		}
	}
	_, err := s.RawDB().Exec(`UPDATE topics SET parent_topic_id = ? WHERE id = ?`, t2, t1)
	if err == nil {
		t.Fatal("expected trigger to refuse non-batch parent")
	}
	if !strings.Contains(err.Error(), "anchor kind = batch") {
		t.Errorf("unexpected trigger message: %v", err)
	}
}

func TestTrigger_topicsBatchNoGrandparent_fires(t *testing.T) {
	s, _ := newTestStore(t)
	if err := s.UpsertUser(User{ID: "u", DisplayName: "U"}); err != nil {
		t.Fatal(err)
	}
	// A batch-anchored topic must not have a parent_topic_id. Seed a valid
	// batch parent (parent_topic_id NULL), then try to attach it to another
	// batch parent — the trigger must refuse.
	batchAnchor, _ := MarshalAnchor(BatchAnchor{})
	outer := uuid.NewString()
	inner := uuid.NewString()
	for _, id := range []string{outer, inner} {
		if err := s.InsertTopic(NewTopic{
			ID: id, SourcePath: "doc.md", Anchor: batchAnchor, CreatedBy: "u",
		}); err != nil {
			t.Fatal(err)
		}
	}
	_, err := s.RawDB().Exec(`UPDATE topics SET parent_topic_id = ? WHERE id = ?`, outer, inner)
	if err == nil {
		t.Fatal("expected trigger to refuse a batch attached to another batch")
	}
	if !strings.Contains(err.Error(), "batch parent cannot itself be a child") {
		t.Errorf("unexpected trigger message: %v", err)
	}
}

If newTestStore doesn't exist as a shared helper, write a small one at the top of migrate_test.go that opens a fresh Store against a temp DB.

• Step 6: Run the new trigger tests + full collab suite

go test ./internal/collab/ -run 'TestTrigger_' -v
go test ./internal/collab/ -v

Expected: PASS. If any pre-existing test breaks because it asserted against the old schema's column order, fix the assertion to use column names rather than positional indexes.

• Step 7: Commit

git add internal/collab/migrations/006_batched_incorporation.sql internal/collab/migrate_test.go
git commit -m "collab: migration 006 — add parent_topic_id, drop composite FK, install batch triggers"

Task 2: Add BatchAnchor to anchor.go

Files:

• Modify: internal/collab/anchor.go
• Test: internal/collab/anchor_test.go

A batch parent's anchor is {"kind": "batch"} with no payload — same shape as GlobalAnchor.

• Step 1: Add a failing test

Append to internal/collab/anchor_test.go:

func TestBatchAnchor_marshalUnmarshalRoundtrip(t *testing.T) {
	raw, err := MarshalAnchor(BatchAnchor{})
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	if got := string(raw); got != `{"kind":"batch"}` {
		t.Fatalf("marshal: got %s", got)
	}
	got, err := UnmarshalAnchor(raw)
	if err != nil {
		t.Fatalf("unmarshal: %v", err)
	}
	if _, ok := got.(BatchAnchor); !ok {
		t.Fatalf("expected BatchAnchor, got %T", got)
	}
}

• Step 2: Run test to verify it fails

go test ./internal/collab/ -run TestBatchAnchor -v

Expected: FAIL (BatchAnchor undefined).

• Step 3: Add the type, kind, and the marshal/unmarshal/validate branches

In internal/collab/anchor.go, add:

type BatchAnchor struct{}

func (BatchAnchor) AnchorKind() string { return "batch" }

In MarshalAnchor's switch, add:

case BatchAnchor:
    return json.Marshal(struct {
        Kind string `json:"kind"`
    }{Kind: v.AnchorKind()})

In UnmarshalAnchor's switch, add:

case "batch":
    var v struct {
        Kind string `json:"kind"`
    }
    if err := decode(&v); err != nil {
        return nil, fmt.Errorf("collab anchor: decode batch: %w", err)
    }
    return BatchAnchor{}, nil

In ValidateAnchor's switch, add:

case BatchAnchor:
    return nil

• Step 4: Run test to verify it passes

go test ./internal/collab/ -run TestBatchAnchor -v

Expected: PASS.

• Step 5: Run the full anchor test file

go test ./internal/collab/ -run TestAnchor -v

Expected: all anchor tests pass.

• Step 6: Commit

git add internal/collab/anchor.go internal/collab/anchor_test.go
git commit -m "collab: add BatchAnchor for batch-parent Topics"

Task 3: Store helpers — CreateBatch, BatchParentForTopic, BatchChildrenOfParent, IsTopicBatched, DiscardBatch

Files:

• Create: internal/collab/batch.go
• Create: internal/collab/batch_test.go

These helpers wrap the SQLite operations that POST /api/batches, handleDiscardTopic, and the realtime / read paths will need.

• Step 1: Define the API surface in a failing test

Create internal/collab/batch_test.go:

package collab

import (
	"path/filepath"
	"testing"

	"github.com/google/uuid"
)

// helper: open a store backed by a temp DB and seed one user + two open
// Topics on the same source. Returns (store, user_id, child_a_id, child_b_id, source_path).
func newBatchTestEnv(t *testing.T) (*Store, string, string, string, string) {
	t.Helper()
	s, err := Open(Config{Path: filepath.Join(t.TempDir(), "collab.db")})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { s.Close() })
	if err := s.UpsertUser(User{ID: "daniel@getorcha.com", DisplayName: "Daniel"}); err != nil {
		t.Fatal(err)
	}
	src := "doc.md"
	anchor, _ := MarshalAnchor(GlobalAnchor{})
	a := uuid.NewString()
	b := uuid.NewString()
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: a, SourcePath: src, Anchor: anchor, CreatedBy: "daniel@getorcha.com",
		FirstMessageID: uuid.NewString(), FirstMessageBody: "child a",
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: b, SourcePath: src, Anchor: anchor, CreatedBy: "daniel@getorcha.com",
		FirstMessageID: uuid.NewString(), FirstMessageBody: "child b",
	}); err != nil {
		t.Fatal(err)
	}
	return s, "daniel@getorcha.com", a, b, src
}

func TestCreateBatch_attachesChildrenToNewParent(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID:  parentID,
		SourcePath:     src,
		ChildTopicIDs:  []string{a, b},
		CreatedBy:      user,
	}); err != nil {
		t.Fatalf("CreateBatch: %v", err)
	}

	// Parent exists with anchor.kind = batch and parent_topic_id NULL.
	var anchor string
	if err := s.RawDB().QueryRow(
		`SELECT anchor FROM topics WHERE id = ?`, parentID,
	).Scan(&anchor); err != nil {
		t.Fatal(err)
	}
	if anchor != `{"kind":"batch"}` {
		t.Fatalf("parent anchor: %s", anchor)
	}

	// Both children have parent_topic_id = parentID.
	for _, child := range []string{a, b} {
		var got string
		if err := s.RawDB().QueryRow(
			`SELECT parent_topic_id FROM topics WHERE id = ?`, child,
		).Scan(&got); err != nil {
			t.Fatal(err)
		}
		if got != parentID {
			t.Fatalf("child %s parent_topic_id: got %s want %s", child, got, parentID)
		}
	}
}

func TestCreateBatch_rejectsAlreadyBatchedChild(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	firstParent := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: firstParent, SourcePath: src, ChildTopicIDs: []string{a, b}, CreatedBy: user,
	}); err != nil {
		t.Fatal(err)
	}
	secondParent := uuid.NewString()
	err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: secondParent, SourcePath: src, ChildTopicIDs: []string{a}, CreatedBy: user,
	})
	if err == nil {
		t.Fatal("expected error when re-batching an attached child")
	}
}

func TestCreateBatch_rejectsCrossSource(t *testing.T) {
	s, user, a, b, _ := newBatchTestEnv(t)
	anchor, _ := MarshalAnchor(GlobalAnchor{})
	other := uuid.NewString()
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: other, SourcePath: "other.md", Anchor: anchor, CreatedBy: user,
		FirstMessageID: uuid.NewString(), FirstMessageBody: "x",
	}); err != nil {
		t.Fatal(err)
	}
	_ = b
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: parentID, SourcePath: "doc.md", ChildTopicIDs: []string{a, other}, CreatedBy: user,
	}); err == nil {
		t.Fatal("expected error when children span sources")
	}
}

func TestBatchParentForTopic_andBatchChildren(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: parentID, SourcePath: src, ChildTopicIDs: []string{a, b}, CreatedBy: user,
	}); err != nil {
		t.Fatal(err)
	}
	got, err := s.BatchParentForTopic(a)
	if err != nil {
		t.Fatal(err)
	}
	if got != parentID {
		t.Fatalf("parent for child a: got %s want %s", got, parentID)
	}
	children, err := s.BatchChildrenOfParent(parentID)
	if err != nil {
		t.Fatal(err)
	}
	if len(children) != 2 {
		t.Fatalf("children: got %d want 2", len(children))
	}
}

func TestIsTopicBatched(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	got, err := s.IsTopicBatched(a)
	if err != nil || got {
		t.Fatalf("unbatched topic: got=%v err=%v", got, err)
	}
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: parentID, SourcePath: src, ChildTopicIDs: []string{a, b}, CreatedBy: user,
	}); err != nil {
		t.Fatal(err)
	}
	got, err = s.IsTopicBatched(a)
	if err != nil || !got {
		t.Fatalf("batched child: got=%v err=%v", got, err)
	}
}

func TestDiscardBatch_clearsChildrenAndWritesAuditMessage(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: parentID, SourcePath: src, ChildTopicIDs: []string{a, b}, CreatedBy: user,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DiscardBatch(DiscardBatchInput{
		ParentTopicID: parentID, ByUser: user, Reason: "reformulating",
	}); err != nil {
		t.Fatal(err)
	}
	// Parent is discarded.
	var discardedAt sql.NullInt64
	if err := s.RawDB().QueryRow(
		`SELECT discarded_at FROM topics WHERE id = ?`, parentID,
	).Scan(&discardedAt); err != nil {
		t.Fatal(err)
	}
	if !discardedAt.Valid {
		t.Fatal("parent not discarded")
	}
	// Children have parent_topic_id cleared and are still open.
	for _, child := range []string{a, b} {
		var p sql.NullString
		var inc, disc sql.NullInt64
		if err := s.RawDB().QueryRow(
			`SELECT parent_topic_id, incorporated_at, discarded_at FROM topics WHERE id = ?`, child,
		).Scan(&p, &inc, &disc); err != nil {
			t.Fatal(err)
		}
		if p.Valid {
			t.Fatalf("child %s still attached: parent=%s", child, p.String)
		}
		if inc.Valid || disc.Valid {
			t.Fatalf("child %s should still be open", child)
		}
	}
	// An audit message exists in the parent's thread containing both child IDs.
	msgs, err := s.ListMessages(parentID)
	if err != nil {
		t.Fatal(err)
	}
	var found bool
	for _, m := range msgs {
		if m.Kind == "human" && strings.Contains(m.Body, a) && strings.Contains(m.Body, b) {
			found = true
			break
		}
	}
	if !found {
		t.Fatal("audit message missing child IDs")
	}
}

Add the necessary imports (database/sql, strings, github.com/google/uuid).

• Step 2: Run tests to verify they fail

go test ./internal/collab/ -run TestCreateBatch -v

Expected: FAIL (CreateBatch undefined).

• Step 3: Implement internal/collab/batch.go

package collab

import (
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"
)

// ErrChildAlreadyBatched is returned when CreateBatch finds one of the
// requested children is already attached to an open parent.
var ErrChildAlreadyBatched = errors.New("child already batched")

// ErrCrossSourceBatch is returned when children span more than one source_path.
var ErrCrossSourceBatch = errors.New("batch children must share a source_path")

// ErrBatchMinimum is returned when fewer than 2 children are requested.
var ErrBatchMinimum = errors.New("batch requires at least 2 children")

// ErrNotBatchParent is returned when a caller asks for batch operations on a
// non-batch Topic.
var ErrNotBatchParent = errors.New("topic is not a batch parent")

// CreateBatchInput is the input to CreateBatch. The parent's id is provided by
// the caller so the HTTP handler can return it in the response without a
// follow-up read; SourcePath is the same as the children's. CreatedBy is the
// FK for both the parent topic row and the audit message it writes on abandon.
type CreateBatchInput struct {
	ParentTopicID string
	SourcePath    string
	ChildTopicIDs []string
	CreatedBy     string
}

// CreateBatch atomically:
//  1. Validates the input (≥2 children, all open, all on the same source,
//     none already batched).
//  2. Inserts a batch parent Topic with anchor.kind = "batch".
//  3. Sets parent_topic_id on every child.
//
// All three steps happen inside a single funnel turn so concurrent batch
// creation can't double-attach a child.
func (s *Store) CreateBatch(in CreateBatchInput) error {
	if in.ParentTopicID == "" || in.SourcePath == "" || in.CreatedBy == "" {
		return fmt.Errorf("collab.CreateBatch: required fields missing")
	}
	if len(in.ChildTopicIDs) < 2 {
		return fmt.Errorf("collab.CreateBatch: %w", ErrBatchMinimum)
	}
	if _, err := ValidateSourcePath(in.SourcePath); err != nil {
		return fmt.Errorf("collab.CreateBatch: %w", err)
	}
	anchor, err := MarshalAnchor(BatchAnchor{})
	if err != nil {
		return fmt.Errorf("collab.CreateBatch: marshal anchor: %w", err)
	}

	return s.send(func(db *sql.DB) error {
		tx, err := db.Begin()
		if err != nil {
			return err
		}
		// Validate each child: open, on the requested source, not batched.
		for _, child := range in.ChildTopicIDs {
			var (
				src string
				parent sql.NullString
				inc, disc sql.NullInt64
			)
			if err := tx.QueryRow(
				`SELECT source_path, parent_topic_id, incorporated_at, discarded_at
				   FROM topics WHERE id = ?`, child,
			).Scan(&src, &parent, &inc, &disc); err != nil {
				_ = tx.Rollback()
				if errors.Is(err, sql.ErrNoRows) {
					return fmt.Errorf("collab.CreateBatch: child %s not found", child)
				}
				return err
			}
			if src != in.SourcePath {
				_ = tx.Rollback()
				return fmt.Errorf("collab.CreateBatch: %w (child %s on %s, want %s)",
					ErrCrossSourceBatch, child, src, in.SourcePath)
			}
			if inc.Valid || disc.Valid {
				_ = tx.Rollback()
				return fmt.Errorf("collab.CreateBatch: child %s: %w", child, ErrTopicAlreadyTerminal)
			}
			if parent.Valid {
				_ = tx.Rollback()
				return fmt.Errorf("collab.CreateBatch: %w (child %s already attached to %s)",
					ErrChildAlreadyBatched, child, parent.String)
			}
		}

		// Insert the parent.
		if _, err := tx.Exec(
			`INSERT INTO topics(id, source_path, anchor, created_at, created_by, updated_at)
			 VALUES (?, ?, ?, unixepoch(), ?, unixepoch())`,
			in.ParentTopicID, in.SourcePath, string(anchor), in.CreatedBy,
		); err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("insert parent: %w", err)
		}

		// Attach each child. The topics_parent_must_be_batch trigger runs on
		// each UPDATE and refuses if the parent doesn't carry anchor.kind=batch.
		for _, child := range in.ChildTopicIDs {
			res, err := tx.Exec(
				`UPDATE topics SET parent_topic_id = ?, updated_at = unixepoch()
				  WHERE id = ?
				    AND parent_topic_id IS NULL
				    AND incorporated_at IS NULL
				    AND discarded_at IS NULL`,
				in.ParentTopicID, child,
			)
			if err != nil {
				_ = tx.Rollback()
				return fmt.Errorf("attach child %s: %w", child, err)
			}
			n, err := res.RowsAffected()
			if err != nil {
				_ = tx.Rollback()
				return err
			}
			if n != 1 {
				// Lost a race after the pre-check above: another batch claimed
				// the child between SELECT and UPDATE.
				_ = tx.Rollback()
				return fmt.Errorf("collab.CreateBatch: %w (child %s claimed concurrently)",
					ErrChildAlreadyBatched, child)
			}
		}

		return tx.Commit()
	})
}

// BatchParentForTopic returns the parent_topic_id for the given topic, or ""
// if the topic is not batched. Returns sql.ErrNoRows for unknown topic IDs.
func (s *Store) BatchParentForTopic(topicID string) (string, error) {
	var parent sql.NullString
	if err := s.db.QueryRow(
		`SELECT parent_topic_id FROM topics WHERE id = ?`, topicID,
	).Scan(&parent); err != nil {
		return "", err
	}
	if !parent.Valid {
		return "", nil
	}
	return parent.String, nil
}

// IsTopicBatched is BatchParentForTopic compressed to a bool.
func (s *Store) IsTopicBatched(topicID string) (bool, error) {
	parent, err := s.BatchParentForTopic(topicID)
	if err != nil {
		return false, err
	}
	return parent != "", nil
}

// IsBatchParent returns true when the given topic has anchor.kind = "batch".
// Used by handler_proposals.go (routing discard / incorporate) and the agent
// service (post-job invariant generalisation).
func (s *Store) IsBatchParent(topicID string) (bool, error) {
	var raw string
	err := s.db.QueryRow(`SELECT anchor FROM topics WHERE id = ?`, topicID).Scan(&raw)
	if err != nil {
		return false, err
	}
	a, err := UnmarshalAnchor([]byte(raw))
	if err != nil {
		return false, err
	}
	_, ok := a.(BatchAnchor)
	return ok, nil
}

// BatchChildrenOfParent returns the open child topic IDs attached to parentID,
// sorted ascending by id. Discarded/incorporated rows whose parent_topic_id
// is non-null (post-incorporation children) are also returned — callers that
// only want open children must filter.
func (s *Store) BatchChildrenOfParent(parentID string) ([]string, error) {
	rows, err := s.db.Query(
		`SELECT id FROM topics WHERE parent_topic_id = ? ORDER BY id`,
		parentID,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		out = append(out, id)
	}
	return out, rows.Err()
}

// DiscardBatchInput drives DiscardBatch.
type DiscardBatchInput struct {
	ParentTopicID string
	ByUser        string
	Reason        string // optional; combined with the children-ID list into the audit message
}

// DiscardBatch flips the parent to discarded, clears parent_topic_id on every
// child, and writes a kind='human' audit message in the parent's thread that
// enumerates the attached children at abandon time. Returns the discarded_at
// timestamp.
func (s *Store) DiscardBatch(in DiscardBatchInput) (int64, error) {
	if in.ParentTopicID == "" || in.ByUser == "" {
		return 0, fmt.Errorf("collab.DiscardBatch: required fields missing")
	}
	now := time.Now().Unix()
	err := s.send(func(db *sql.DB) error {
		tx, err := db.Begin()
		if err != nil {
			return err
		}
		// Confirm the target is an open batch parent.
		var anchor string
		var inc, disc sql.NullInt64
		if err := tx.QueryRow(
			`SELECT anchor, incorporated_at, discarded_at FROM topics WHERE id = ?`,
			in.ParentTopicID,
		).Scan(&anchor, &inc, &disc); err != nil {
			_ = tx.Rollback()
			return err
		}
		a, err := UnmarshalAnchor([]byte(anchor))
		if err != nil {
			_ = tx.Rollback()
			return err
		}
		if _, ok := a.(BatchAnchor); !ok {
			_ = tx.Rollback()
			return fmt.Errorf("collab.DiscardBatch: %w", ErrNotBatchParent)
		}
		if inc.Valid || disc.Valid {
			_ = tx.Rollback()
			return fmt.Errorf("collab.DiscardBatch: %w", ErrTopicAlreadyTerminal)
		}

		// Collect children for the audit message + the detach.
		rows, err := tx.Query(
			`SELECT id FROM topics WHERE parent_topic_id = ? ORDER BY id`,
			in.ParentTopicID,
		)
		if err != nil {
			_ = tx.Rollback()
			return err
		}
		var children []string
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				_ = tx.Rollback()
				return err
			}
			children = append(children, id)
		}
		rows.Close()

		// Build the audit message.
		var sb strings.Builder
		sb.WriteString("Batch abandoned. Children at abandon: ")
		sb.WriteString(strings.Join(children, ", "))
		if reason := strings.TrimSpace(in.Reason); reason != "" {
			sb.WriteString(". Reason: ")
			sb.WriteString(reason)
		}
		auditBody := sb.String()

		// Allocate the next sequence in the parent's thread and insert the audit row.
		var nextSeq int
		if err := tx.QueryRow(
			`SELECT COALESCE(MAX(sequence), 0) + 1 FROM topic_messages WHERE topic_id = ?`,
			in.ParentTopicID,
		).Scan(&nextSeq); err != nil {
			_ = tx.Rollback()
			return err
		}
		msgID := uuid.NewString()
		by := in.ByUser
		if _, err := tx.Exec(
			`INSERT INTO topic_messages(
			   id, topic_id, kind, body, author_user_id,
			   proposal_id, sequence, created_at
			 ) VALUES (?, ?, 'human', ?, ?, NULL, ?, ?)`,
			msgID, in.ParentTopicID, auditBody, by, nextSeq, now,
		); err != nil {
			_ = tx.Rollback()
			return err
		}

		// Clear parent_topic_id on every child.
		if _, err := tx.Exec(
			`UPDATE topics SET parent_topic_id = NULL, updated_at = ?
			  WHERE parent_topic_id = ?
			    AND incorporated_at IS NULL
			    AND discarded_at IS NULL`,
			now, in.ParentTopicID,
		); err != nil {
			_ = tx.Rollback()
			return err
		}

		// Discard the parent.
		res, err := tx.Exec(
			`UPDATE topics
			    SET discarded_at = ?, discarded_by = ?, updated_at = ?
			  WHERE id = ?
			    AND incorporated_at IS NULL
			    AND discarded_at IS NULL`,
			now, in.ByUser, now, in.ParentTopicID,
		)
		if err != nil {
			_ = tx.Rollback()
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			_ = tx.Rollback()
			return err
		}
		if n != 1 {
			_ = tx.Rollback()
			return ErrTopicAlreadyTerminal
		}

		return tx.Commit()
	})
	return now, err
}

• Step 4: Run the new tests

go test ./internal/collab/ -run 'TestCreateBatch|TestBatchParent|TestIsTopicBatched|TestDiscardBatch' -v

Expected: all pass.

• Step 5: Run the full collab suite

go test ./internal/collab/ -v

Expected: all pass.

• Step 6: Commit

git add internal/collab/batch.go internal/collab/batch_test.go
git commit -m "collab: store helpers for batch creation, lookup, and abandon"

Task 4: Add parent_topic_id to TopicSummary and surface it from list reads

Files:

• Modify: internal/collab/reader.go
• Modify: internal/collab/reader_test.go

The chrome sidebar needs to know which Topics are children of which batch parents. The read API returns this on the existing list endpoint.

• Step 1: Extend TopicSummary

In internal/collab/reader.go:

type TopicSummary struct {
	ID                  string          `json:"id"`
	SourcePath          string          `json:"source_path"`
	Anchor              json.RawMessage `json:"anchor"`
	ParentTopicID       *string         `json:"parent_topic_id,omitempty"`
	CreatedBy           string          `json:"created_by"`
	CreatedAt           int64           `json:"created_at"`
	UpdatedAt           int64           `json:"updated_at"`
	FirstMessagePreview string          `json:"first_message_preview"`
	MessageCount        int             `json:"message_count"`
}

• Step 2: Update the SELECT queries and scanTopicSummaries to include parent_topic_id

In ListOpenTopicsForSource and ListOpenTopicsForSourceExcluding, change the SELECT list to include t.parent_topic_id:

`SELECT t.id, t.source_path, t.anchor, t.parent_topic_id, t.created_by, t.created_at, t.updated_at,
        COALESCE((SELECT m.body FROM topic_messages m WHERE m.topic_id = t.id ORDER BY m.sequence LIMIT 1), ''),
        (SELECT COUNT(*) FROM topic_messages m WHERE m.topic_id = t.id)
   FROM topics t
  WHERE …`

And update scanTopicSummaries:

func scanTopicSummaries(rows *sql.Rows) ([]TopicSummary, error) {
	out := []TopicSummary{}
	for rows.Next() {
		var raw string
		var parent sql.NullString
		var t TopicSummary
		if err := rows.Scan(
			&t.ID, &t.SourcePath, &raw, &parent, &t.CreatedBy, &t.CreatedAt, &t.UpdatedAt,
			&t.FirstMessagePreview, &t.MessageCount,
		); err != nil {
			return nil, err
		}
		if _, err := UnmarshalAnchor(json.RawMessage(raw)); err != nil {
			return nil, fmt.Errorf("scanTopicSummaries: topic %s: %w", t.ID, err)
		}
		t.Anchor = json.RawMessage(raw)
		if parent.Valid {
			v := parent.String
			t.ParentTopicID = &v
		}
		out = append(out, t)
	}
	return out, rows.Err()
}

• Step 3: Add a focused test

Append to internal/collab/reader_test.go:

func TestListOpenTopicsForSource_returnsParentTopicID(t *testing.T) {
	s, user, a, b, src := newBatchTestEnv(t)
	parentID := uuid.NewString()
	if err := s.CreateBatch(CreateBatchInput{
		ParentTopicID: parentID, SourcePath: src,
		ChildTopicIDs: []string{a, b}, CreatedBy: user,
	}); err != nil {
		t.Fatal(err)
	}
	rows, err := s.ListOpenTopicsForSource(src)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 3 {
		t.Fatalf("expected 3 topics (parent + 2 children), got %d", len(rows))
	}
	byID := make(map[string]TopicSummary, len(rows))
	for _, r := range rows {
		byID[r.ID] = r
	}
	if byID[parentID].ParentTopicID != nil {
		t.Errorf("parent should have nil ParentTopicID")
	}
	for _, child := range []string{a, b} {
		got := byID[child].ParentTopicID
		if got == nil || *got != parentID {
			t.Errorf("child %s: ParentTopicID = %v want %s", child, got, parentID)
		}
	}
}

• Step 4: Run and verify

go test ./internal/collab/ -run TestListOpenTopicsForSource_returnsParentTopicID -v
go test ./internal/collab/ -v

Expected: PASS; no regressions.

• Step 5: Commit

git add internal/collab/reader.go internal/collab/reader_test.go
git commit -m "collab: expose parent_topic_id on TopicSummary"

Task 5: ClaimIncorporation and freshness exclude batched + batch parents

Files:

• Modify: internal/collab/mutators.go (ClaimIncorporation query)
• Modify: internal/collab/freshness.go (openNonGlobalTopicIDsExcluding query)
• Modify: internal/collab/freshness_test.go (extend) and internal/collab/mutators_test.go (extend)

Both functions currently scan "open non-global Topics" to enforce marker presence. They must additionally exclude (a) Topics already batched as children of an open parent, and (b) Topics that are themselves batch parents. The first set is "being incorporated together"; the second has no marker by definition.

• Step 1: Write a failing freshness test

Append to internal/collab/freshness_test.go:

func TestProposalFreshness_excludesBatchedChildrenAndBatchParents(t *testing.T) {
	root := t.TempDir()
	src := "doc.md"
	if err := os.WriteFile(filepath.Join(root, src), []byte("hello"), 0o644); err != nil {
		t.Fatal(err)
	}
	repoSHA := mustGitInit(t, root) // helper used elsewhere in this package; see existing tests
	_ = repoSHA

	s, _ := newTestStore(t) // existing helper
	if err := s.UpsertUser(User{ID: "u", DisplayName: "U"}); err != nil {
		t.Fatal(err)
	}

	// Topic A: open, anchored, has a marker in the proposed source.
	a := uuid.NewString()
	preMarkerAnchor, _ := MarshalAnchor(PreMarkerAnchor{
		SourceSHA: strings.Repeat("0", 40), Start: 0, End: 5, Quote: "hello",
	})
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: a, SourcePath: src, Anchor: preMarkerAnchor, CreatedBy: "u",
		FirstMessageID: uuid.NewString(), FirstMessageBody: "x",
	}); err != nil {
		t.Fatal(err)
	}

	// Topic P (the parent of a different batch) — has no marker in the proposed
	// source. With the change in this task, it must NOT cause the freshness check
	// to fail.
	p := uuid.NewString()
	batchAnchor, _ := MarshalAnchor(BatchAnchor{})
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: p, SourcePath: src, Anchor: batchAnchor, CreatedBy: "u",
		FirstMessageID: uuid.NewString(), FirstMessageBody: "y",
	}); err != nil {
		t.Fatal(err)
	}

	// Topic C — a batched child of P, with no marker in the proposed source.
	// Must also not cause freshness to fail.
	c := uuid.NewString()
	if err := s.InsertTopicWithFirstMessage(NewTopicWithFirstMessage{
		TopicID: c, SourcePath: src, Anchor: preMarkerAnchor, CreatedBy: "u",
		FirstMessageID: uuid.NewString(), FirstMessageBody: "z",
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.RawDB().Exec(
		`UPDATE topics SET parent_topic_id = ? WHERE id = ?`, p, c,
	); err != nil {
		t.Fatal(err)
	}

	// Proposal on Topic A whose proposed_source carries A's marker — no markers
	// for P or C.
	proposalID := uuid.NewString()
	if _, err := s.InsertProposal(NewProposal{
		ID: proposalID, TopicID: a, RevisionNumber: 1,
		ProposedSource: `<span data-orcha-anchor="` + a + `">hello</span>`,
		BaseSourceSHA:  mustHashFile(t, root, src),
	}); err != nil {
		t.Fatal(err)
	}

	fr, err := ProposalFreshness(s, root, proposalID)
	if err != nil {
		t.Fatal(err)
	}
	if !fr.Fresh {
		t.Fatalf("expected fresh; got stale_reasons=%v missing=%v", fr.StaleReasons, fr.MissingTopicIDs)
	}
}

If mustGitInit / mustHashFile / newTestStore helpers don't already exist in the package, port the ones used by existing tests in freshness_test.go — do not reinvent.

• Step 2: Run the test and verify it fails

go test ./internal/collab/ -run TestProposalFreshness_excludesBatchedChildrenAndBatchParents -v

Expected: FAIL — freshness sees C as an open non-global Topic missing its marker.

• Step 3: Update openNonGlobalTopicIDsExcluding in freshness.go

Change the SQL filter to add two new exclusions:

`SELECT id FROM topics
  WHERE source_path = ?
    AND incorporated_at IS NULL
    AND discarded_at IS NULL
    AND id != ?
    AND json_extract(anchor, '$.kind') != 'global'
    AND json_extract(anchor, '$.kind') != 'batch'
    AND parent_topic_id IS NULL
  ORDER BY id`

• Step 4: Apply the same change to ClaimIncorporation in mutators.go

rows, err := tx.Query(
    `SELECT id, json_extract(anchor, '$.kind')
       FROM topics
      WHERE source_path = ?
        AND incorporated_at IS NULL
        AND discarded_at IS NULL
        AND id != ?
        AND json_extract(anchor, '$.kind') != 'global'
        AND json_extract(anchor, '$.kind') != 'batch'
        AND parent_topic_id IS NULL
      ORDER BY id`,
    in.Attempt.SourcePath, in.Attempt.TopicID,
)

• Step 5: Run the freshness test and the full collab suite

go test ./internal/collab/ -run TestProposalFreshness_excludesBatchedChildrenAndBatchParents -v
go test ./internal/collab/ -v

Expected: PASS.

• Step 6: Commit

git add internal/collab/freshness.go internal/collab/freshness_test.go internal/collab/mutators.go
git commit -m "collab: exclude batched children and batch parents from marker-set checks"