fix(batches): resolve batch ids beyond the 100-batch cap (#825)

## Summary
- Add `conductor.find_batch_by_prefix` (SQL `LIKE 'prefix%'`, no cap, escapes LIKE metachars) so `cf work batch status/stop/resume/follow` can resolve batches beyond the newest 100 — same fix class as #743/#824 for tasks.
- Replace the four `list_batches(limit=100)` + Python `startswith` resolution sites with the uncapped SQL lookup.
- Add `len(matching) > 1` → "Multiple batches match" + `typer.Exit(1)` ambiguity guard to all four commands (mutating `stop`/`resume` previously acted silently / warned-then-proceeded), matching the task-side pattern — closes the widened-collision-surface gap the cross-family review flagged.
- Regression tests: 6 core tests for `find_batch_by_prefix` + a parametrized `CliRunner` test proving each command rejects an ambiguous prefix.

## Validation
- Tests: All passing (10/10 locally + full CI green)
- Lint: Clean (ruff + mypy)
- Test mutation check: Passed — ambiguity test fails without the guard
- Internal review: Completed (advisory, Claude, 2 rounds)
- Cross-family review: opencode (zai/glm-5.2) — initial REQUEST_CHANGES (Critical on batch_stop) resolved; GLM CI re-review + Claude follow-up confirm no new defects
- Final feedback triage: cutoff 2026-07-20T18:48:39Z, all items triaged (2 non-blocking deferred with justification)
- Demo: All acceptance criteria verified with real-CLI outcome evidence on a 156-batch workspace

Closes #825

Author: frankbria

codeframe/cli/app.py

@@ -4017,14 +4017,19 @@ def batch_status(
 
         if batch_id:
             # Show specific batch
-            # Find by partial ID
-            all_batches = conductor.list_batches(workspace, limit=100)
-            matching = [b for b in all_batches if b.id.startswith(batch_id)]
+            # Find by partial ID (SQL LIKE — no cap, resolves batches beyond 100)
+            matching = conductor.find_batch_by_prefix(workspace, batch_id)
 
             if not matching:
                 console.print(f"[red]Error:[/red] No batch found matching '{batch_id}'")
                 raise typer.Exit(1)
 
+            if len(matching) > 1:
+                console.print(f"[red]Error:[/red] Multiple batches match '{batch_id}':")
+                for b in matching[:5]:
+                    console.print(f"  {b.id[:8]} ({b.status.value})")
+                raise typer.Exit(1)
+
             batch = matching[0]
 
             # Status color
@@ -4162,14 +4167,19 @@ def batch_stop(
     try:
         workspace = get_workspace(path)
 
-        # Find by partial ID
-        all_batches = conductor.list_batches(workspace, limit=100)
-        matching = [b for b in all_batches if b.id.startswith(batch_id)]
+        # Find by partial ID (SQL LIKE — no cap, resolves batches beyond 100)
+        matching = conductor.find_batch_by_prefix(workspace, batch_id)
 
         if not matching:
             console.print(f"[red]Error:[/red] No batch found matching '{batch_id}'")
             raise typer.Exit(1)
 
+        if len(matching) > 1:
+            console.print(f"[red]Error:[/red] Multiple batches match '{batch_id}':")
+            for b in matching[:5]:
+                console.print(f"  {b.id[:8]} ({b.status.value})")
+            raise typer.Exit(1)
+
         batch = matching[0]
 
         if batch.status.value not in ("PENDING", "RUNNING"):
@@ -4231,19 +4241,18 @@ def batch_resume(
     try:
         workspace = get_workspace(path)
 
-        # Find by partial ID
-        all_batches = conductor.list_batches(workspace, limit=100)
-        matching = [b for b in all_batches if b.id.startswith(batch_id)]
+        # Find by partial ID (SQL LIKE — no cap, resolves batches beyond 100)
+        matching = conductor.find_batch_by_prefix(workspace, batch_id)
 
         if not matching:
             console.print(f"[red]Error:[/red] No batch found matching '{batch_id}'")
             raise typer.Exit(1)
 
         if len(matching) > 1:
-            console.print(f"[yellow]Warning:[/yellow] Multiple batches match '{batch_id}':")
+            console.print(f"[red]Error:[/red] Multiple batches match '{batch_id}':")
             for b in matching[:5]:
-                console.print(f"  - {b.id[:8]} ({b.status.value})")
-            console.print("Using the most recent match.")
+                console.print(f"  {b.id[:8]} ({b.status.value})")
+            raise typer.Exit(1)
 
         batch = matching[0]
 
@@ -4430,14 +4439,19 @@ def _update_progress(progress: BatchProgress, event: events.Event) -> None:
     try:
         workspace = get_workspace(path)
 
-        # Find batch by partial ID
-        all_batches = conductor.list_batches(workspace, limit=100)
-        matching = [b for b in all_batches if b.id.startswith(batch_id)]
+        # Find batch by partial ID (SQL LIKE — no cap, resolves batches beyond 100)
+        matching = conductor.find_batch_by_prefix(workspace, batch_id)
 
         if not matching:
             console.print(f"[red]Error:[/red] No batch found matching '{batch_id}'")
             raise typer.Exit(1)
 
+        if len(matching) > 1:
+            console.print(f"[red]Error:[/red] Multiple batches match '{batch_id}':")
+            for b in matching[:5]:
+                console.print(f"  {b.id[:8]} ({b.status.value})")
+            raise typer.Exit(1)
+
         batch = matching[0]
         batch_id_short = batch.id[:8]
 

codeframe/core/conductor.py

@@ -827,6 +827,48 @@ def list_batches(
     return [_row_to_batch(row) for row in rows]
 
 
+def find_batch_by_prefix(workspace: Workspace, prefix: str) -> list[BatchRun]:
+    """Resolve batches whose id starts with ``prefix`` (SQL ``LIKE``, no cap).
+
+    The CLI accepts partial batch ids. Doing that in Python over
+    ``list_batches(limit=100)`` meant any batch beyond the cap was unreachable
+    (#825). This resolves at the SQL layer with no ``LIMIT`` so every matching
+    batch is addressable regardless of how many batches exist.
+
+    Returns all matches (possibly >1) so callers can report ambiguity.
+
+    Args:
+        workspace: Workspace to query
+        prefix: Batch id prefix (user-typed; may be partial)
+
+    Returns:
+        List of matching BatchRuns, newest first.
+    """
+    # Batch IDs are UUIDs (no _/%), but the prefix is user-typed — escape LIKE
+    # metachars so a stray wildcard can't over-match the wrong batch.
+    escaped = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    conn = get_db_connection(workspace)
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            SELECT id, workspace_id, task_ids, status, strategy, max_parallel,
+                   on_failure, started_at, completed_at, results, engine,
+                   isolation, stall_timeout_s, stall_action, concurrency_by_status,
+                   llm_provider, llm_model
+            FROM batch_runs
+            WHERE workspace_id = ? AND id LIKE ? ESCAPE '\\'
+            ORDER BY started_at DESC
+            """,
+            (workspace.id, f"{escaped}%"),
+        )
+        rows = cursor.fetchall()
+    finally:
+        conn.close()
+
+    return [_row_to_batch(row) for row in rows]
+
+
 def cancel_batch(workspace: Workspace, batch_id: str) -> BatchRun:
     """Cancel a running batch.
 

tests/core/test_batch_truncation.py

@@ -0,0 +1,151 @@
+"""Regression tests for #825 — CLI batch resolution truncated at 100.
+
+The four ``cf work batch status/stop/resume/follow`` commands resolved a
+partial batch id in Python over ``conductor.list_batches(workspace, limit=100)``,
+so any batch beyond the first 100 was unreachable — the same class of bug as
+#743 for tasks.
+
+Fix: a SQL ``LIKE 'prefix%'`` lookup (``find_batch_by_prefix``) that bypasses
+the cap, mirroring ``tasks.find_by_prefix``.
+"""
+
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+
+from codeframe.core import conductor
+from codeframe.core.conductor import (
+    BatchRun,
+    BatchStatus,
+    OnFailure,
+    _save_batch,
+)
+from codeframe.core.workspace import create_or_load_workspace
+
+pytestmark = pytest.mark.v2
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _make_batch(workspace, *, status: BatchStatus = BatchStatus.PENDING) -> BatchRun:
+    """Insert a minimal BatchRun directly without spawning real workers."""
+    batch = BatchRun(
+        id=str(uuid.uuid4()),
+        workspace_id=workspace.id,
+        task_ids=[],
+        status=status,
+        strategy="serial",
+        max_parallel=1,
+        on_failure=OnFailure.CONTINUE,
+        started_at=_utc_now(),
+        completed_at=None,
+    )
+    _save_batch(workspace, batch)
+    return batch
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    return create_or_load_workspace(tmp_path)
+
+
+@pytest.fixture
+def many_batches(workspace):
+    """150 batches — more than the old hard limit of 100."""
+    created = [_make_batch(workspace) for _ in range(150)]
+    return workspace, created
+
+
+class TestFindBatchByPrefixResolvesBeyond100:
+    def test_resolves_batch_beyond_first_100(self, many_batches):
+        """AC1: a batch sorted past position 100 is resolvable by prefix."""
+        workspace, created = many_batches
+        # list_batches is newest-first; oldest batches (earliest indices) are
+        # the ones a LIMIT 100 would drop from the top-100 window.
+        target = created[0]
+
+        matches = conductor.find_batch_by_prefix(workspace, target.id[:8])
+
+        assert [b.id for b in matches] == [target.id]
+
+    def test_exact_full_id_resolves(self, workspace):
+        """A full UUID prefix still resolves correctly."""
+        batch = _make_batch(workspace)
+        matches = conductor.find_batch_by_prefix(workspace, batch.id)
+        assert len(matches) == 1
+        assert matches[0].id == batch.id
+
+    def test_no_match_returns_empty(self, workspace):
+        _make_batch(workspace)
+        assert conductor.find_batch_by_prefix(workspace, "zzzzzzzz-nope") == []
+
+    def test_ambiguous_prefix_returns_multiple(self, workspace):
+        """Empty prefix (or shared prefix) returns all matching batches."""
+        a = _make_batch(workspace)
+        b = _make_batch(workspace)
+        # Empty prefix → LIKE '%' → matches everything
+        matches = conductor.find_batch_by_prefix(workspace, "")
+        ids = {m.id for m in matches}
+        assert a.id in ids and b.id in ids
+
+    def test_like_metachar_prefix_does_not_over_match(self, workspace):
+        """A user-typed prefix with LIKE wildcards must not over-match.
+
+        Batch IDs are UUIDs (never contain ``_``/``%``), so a prefix that is
+        only a wildcard should match nothing rather than every batch.
+        """
+        _make_batch(workspace)
+        _make_batch(workspace)
+        assert conductor.find_batch_by_prefix(workspace, "%") == []
+        assert conductor.find_batch_by_prefix(workspace, "_") == []
+
+    def test_backslash_in_prefix_does_not_crash(self, workspace):
+        """A backslash in the prefix is escaped and returns empty (no UUID match)."""
+        _make_batch(workspace)
+        assert conductor.find_batch_by_prefix(workspace, "\\") == []
+
+
+class TestCliAmbiguousPrefixIsRejected:
+    """Uncapped resolution widens the collision surface, so every batch CLI
+    command must refuse an ambiguous prefix rather than silently act on the
+    first match (matching the task-side ``Multiple tasks match`` guard). This
+    is the merge-blocking finding from PR #872's cross-family review.
+    """
+
+    def _seed_colliding_pair(self, workspace) -> None:
+        """Two batches sharing the prefix ``dupe`` so ``dupe`` matches both."""
+        for suffix in ("0001", "0002"):
+            _save_batch(
+                workspace,
+                BatchRun(
+                    id=f"dupe{suffix}-1111-1111-1111-111111111111",
+                    workspace_id=workspace.id,
+                    task_ids=[],
+                    status=BatchStatus.PENDING,
+                    strategy="serial",
+                    max_parallel=1,
+                    on_failure=OnFailure.CONTINUE,
+                    started_at=_utc_now(),
+                    completed_at=None,
+                ),
+            )
+
+    @pytest.mark.parametrize("subcommand", ["status", "stop", "resume", "follow"])
+    def test_ambiguous_prefix_errors_and_exits(self, workspace, subcommand):
+        from typer.testing import CliRunner
+
+        from codeframe.cli.app import app
+
+        self._seed_colliding_pair(workspace)
+        runner = CliRunner(env={"COLUMNS": "200", "NO_COLOR": "1"})
+
+        result = runner.invoke(
+            app,
+            ["work", "batch", subcommand, "dupe", "-w", str(workspace.repo_path)],
+        )
+
+        assert result.exit_code == 1, result.output
+        assert "Multiple batches match" in result.output