fix(core): implement workspace grounding and exploration abort guards

bright_vision_core/http_api.py

@@ -22,7 +22,7 @@
 from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
 
@@ -125,9 +125,16 @@ class ModelRouterRequest(BaseModel):
     token_fast_max: int = 4_096
     token_heavy_min: int = 12_000
     keep_alive_fast: int | str = 300
-    keep_alive_heavy: int | str = 0
+    keep_alive_heavy: int | str = -1
     escalate_on_failure: bool = True
 
+    @field_validator("keep_alive_heavy", mode="before")
+    @classmethod
+    def _normalize_heavy_keep_alive(cls, value: int | str | None) -> int | str:
+        from bright_vision_core.model_router import normalize_keep_alive_for_tier
+
+        return normalize_keep_alive_for_tier("heavy", -1 if value is None else value)
+
 
 class CreateSessionRequest(BaseModel):
     workspace: str = Field(..., description="Absolute path to git workspace root")

bright_vision_core/implement_workspace.py

@@ -0,0 +1,225 @@
+"""Ground spec-focus implement turns in on-disk workspace facts (avoid ls loops)."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+
+from bright_vision_core.spec_steering import workspace_lib_missing
+from bright_vision_core.workspace_todos import ChecklistItem
+
+_PATH_IN_CHECKLIST = re.compile(
+    r"(?:`((?:lib|test)/[\w./-]+)`|((?:lib|test)/[\w./-]+))",
+    re.IGNORECASE,
+)
+
+_SNAPSHOT_DIRS = ("lib", "test")
+_MAX_LIST_FILES = 24
+
+
+def list_workspace_test_files(workspace: str | Path, *, limit: int = _MAX_LIST_FILES) -> list[str]:
+    return _list_tree_files(Path(workspace).resolve(), "test", limit=limit)
+
+
+def _list_tree_files(root: Path, subdir: str, *, limit: int = _MAX_LIST_FILES) -> list[str]:
+    base = root / subdir
+    if not base.is_dir():
+        return []
+    out: list[str] = []
+    for path in sorted(base.rglob("*")):
+        if not path.is_file():
+            continue
+        if path.name.startswith("."):
+            continue
+        rel = path.relative_to(root).as_posix()
+        out.append(rel)
+        if len(out) >= limit:
+            break
+    return out
+
+
+def paths_from_checklist_text(text: str) -> list[str]:
+    found: list[str] = []
+    seen: set[str] = set()
+    for match in _PATH_IN_CHECKLIST.finditer(text or ""):
+        raw = (match.group(1) or match.group(2) or "").strip().rstrip("/")
+        if not raw or raw in seen:
+            continue
+        seen.add(raw)
+        found.append(raw)
+    return found
+
+
+def deliverable_paths_exist(workspace: str | Path, paths: list[str]) -> bool:
+    """True when every path is an existing file or non-empty directory."""
+    root = Path(workspace).resolve()
+    if not paths:
+        return False
+    for rel in paths:
+        target = root / rel
+        if target.is_file():
+            continue
+        if target.is_dir() and any(target.iterdir()):
+            continue
+        return False
+    return True
+
+
+def first_open_checklist_item(checklist: list[ChecklistItem]) -> ChecklistItem | None:
+    for entry in checklist:
+        if not entry.done and entry.text.strip():
+            return entry
+    return None
+
+
+def build_workspace_snapshot_lines(workspace: str | Path) -> list[str]:
+    root = Path(workspace).resolve()
+    lines = ["## Workspace snapshot (verified on disk — do **not** ls to rediscover)"]
+    pubspec = root / "pubspec.yaml"
+    if pubspec.is_file():
+        lines.append("- `pubspec.yaml` — present")
+    else:
+        lines.append("- `pubspec.yaml` — **missing**")
+
+    for sub in _SNAPSHOT_DIRS:
+        files = _list_tree_files(root, sub)
+        if not files:
+            lines.append(f"- `{sub}/` — **empty or missing**")
+            continue
+        preview = ", ".join(f"`{f}`" for f in files[:8])
+        extra = f" (+{len(files) - 8} more)" if len(files) > 8 else ""
+        lines.append(f"- `{sub}/` — {len(files)} file(s): {preview}{extra}")
+    return lines
+
+
+def build_implement_next_action_lines(
+    workspace: str | Path,
+    checklist: list[ChecklistItem],
+    *,
+    resume: bool,
+) -> list[str]:
+    lines = ["## Next action (this turn)"]
+    focus = first_open_checklist_item(checklist)
+    if focus is None:
+        lines.append(
+            "All checklist items are marked done. Run project tests if applicable, "
+            "then update the task status — **no ls/Grep exploration**."
+        )
+        return lines
+
+    paths = paths_from_checklist_text(focus.text)
+    lower = focus.text.lower()
+    on_disk = deliverable_paths_exist(workspace, paths) if paths else False
+    test_files = _list_tree_files(Path(workspace).resolve(), "test")
+
+    if ("test" in lower or "verify" in lower) and test_files:
+        target = next((f for f in test_files if "test" in f), test_files[0])
+        lines.append(
+            f"Focus checklist: **{focus.text.strip()}** — test file(s) already on disk."
+        )
+        lines.append(
+            f"1. **ReadRange** `{target}` with `@000` / `000@` once\n"
+            f"2. **EditText** only if tests need fixes\n"
+            f"3. BrightVision runs **`flutter test`** at end of this turn\n"
+            f"4. Mark checklist item done after tests pass\n"
+            f"**Do not** call ls, Grep, GitStatus, or repeat ReadRange on the same file."
+        )
+    elif on_disk and ("test" in lower or "verify" in lower):
+        test_files = _list_tree_files(Path(workspace).resolve(), "test")
+        target = next((f for f in test_files if "test" in f), test_files[0] if test_files else None)
+        if target:
+            lines.append(
+                f"Focus checklist: **{focus.text.strip()}** — deliverable files already exist."
+            )
+            lines.append(
+                f"1. **ReadRange** `{target}` with `@000` / `000@` once\n"
+                f"2. **EditText** only if tests need fixes\n"
+                f"3. BrightVision will run **`flutter test`** when this turn edits test files\n"
+                f"4. Mark checklist item done after tests pass\n"
+                f"**Do not** call ls, Grep, GitStatus, or repeat ReadRange on the same file."
+            )
+        else:
+            lines.append(
+                f"Focus: **{focus.text.strip()}** — create tests with **ContextManager** + "
+                "**ReadRange** + **EditText** (one file). **No ls.**"
+            )
+    elif on_disk:
+        target = paths[0] if paths else "lib/"
+        lines.append(
+            f"Focus checklist: **{focus.text.strip()}** — paths exist on disk (`{target}`)."
+        )
+        lines.append(
+            f"**ReadRange** the target source file, then **EditText** to finish. **No ls.**"
+        )
+    elif workspace_lib_missing(workspace):
+        lines.append(f"Focus checklist: **{focus.text.strip()}**")
+        lines.append(
+            "**ContextManager** to scaffold `lib/` (and `test/` if needed), then **ReadRange** + "
+            "**EditText** on one file. **Do not ls** empty directories."
+        )
+    elif resume:
+        lines.append(f"Focus checklist: **{focus.text.strip()}**")
+        lines.append(
+            "Use **ReadRange** + **EditText** on **one file** for this item. "
+            "**Do not** ls, Grep, or GitStatus — use the workspace snapshot above."
+        )
+    else:
+        lines.append(f"Focus checklist: **{focus.text.strip()}**")
+        lines.append(
+            "Work this item only: **ContextManager** / **ReadRange** / **EditText**. **No ls.**"
+        )
+    return lines
+
+
+def build_implement_workspace_block(
+    workspace: str | Path,
+    checklist: list[ChecklistItem] | None,
+    *,
+    resume: bool,
+) -> str:
+    """Markdown block injected on implement / resume turns."""
+    parts = build_workspace_snapshot_lines(workspace)
+    if checklist:
+        parts.append("")
+        parts.extend(build_implement_next_action_lines(workspace, checklist, resume=resume))
+    parts.append("")
+    parts.append(
+        "**Hard rule:** Do not batch UpdateTodoList JSON with other tool args. "
+        "One tool per call. Do not call **ls** when this snapshot is present."
+    )
+    return "\n".join(parts)
+
+
+def edited_dart_test_files(edited_files: list[str]) -> list[str]:
+    out: list[str] = []
+    for raw in edited_files:
+        rel = raw.replace("\\", "/").lstrip("./")
+        if rel.startswith("test/") and rel.endswith("_test.dart"):
+            out.append(rel)
+    return out
+
+
+def run_flutter_tests(workspace: str | Path, test_paths: list[str]) -> tuple[bool, str]:
+    """Run ``flutter test`` on specific files; return (passed, combined output)."""
+    root = Path(workspace).resolve()
+    if not (root / "pubspec.yaml").is_file():
+        return False, "pubspec.yaml missing — cannot run flutter test"
+    if not test_paths:
+        return False, "no test paths"
+    cmd = ["flutter", "test", *test_paths]
+    try:
+        proc = subprocess.run(
+            cmd,
+            cwd=str(root),
+            capture_output=True,
+            text=True,
+            timeout=300,
+        )
+    except subprocess.TimeoutExpired:
+        return False, "flutter test timed out after 300s"
+    except FileNotFoundError:
+        return False, "flutter not found on PATH"
+    out = (proc.stdout or "") + (proc.stderr or "")
+    tail = out.strip()[-4000:] if out.strip() else "(no output)"
+    return proc.returncode == 0, tail

bright_vision_core/model_router.py

@@ -15,6 +15,12 @@
 
 RouteTier = Literal["fast", "heavy"]
 
+# Heavy tier keep_alive=0 unloads Ollama after every LLM call — agent loops get empty responses.
+def normalize_keep_alive_for_tier(tier: RouteTier, value: int | str) -> int | str:
+    if tier == "heavy" and value in (0, "0"):
+        return -1
+    return value
+
 # Per-file context bump for *display* only (routing uses message_tokens).
 _FILE_TOKEN_PER_FILE = 500
 _FILE_TOKEN_CAP = 2_000
@@ -94,9 +100,12 @@ class ModelRouterConfig:
     token_fast_max: int = 4_096
     token_heavy_min: int = 12_000
     keep_alive_fast: int | str = 300
-    keep_alive_heavy: int | str = 0
+    keep_alive_heavy: int | str = -1
     escalate_on_failure: bool = True
 
+    def __post_init__(self) -> None:
+        self.keep_alive_heavy = normalize_keep_alive_for_tier("heavy", self.keep_alive_heavy)
+
     @classmethod
     def from_payload(cls, raw: dict[str, Any] | None) -> ModelRouterConfig | None:
         if not raw:
@@ -142,7 +151,9 @@ def from_payload(cls, raw: dict[str, Any] | None) -> ModelRouterConfig | None:
             token_fast_max=int(raw.get("token_fast_max") or 4_096),
             token_heavy_min=int(raw.get("token_heavy_min") or 12_000),
             keep_alive_fast=raw.get("keep_alive_fast", 300),
-            keep_alive_heavy=raw.get("keep_alive_heavy", 0),
+            keep_alive_heavy=normalize_keep_alive_for_tier(
+                "heavy", raw.get("keep_alive_heavy", -1)
+            ),
             escalate_on_failure=bool(raw.get("escalate_on_failure", True)),
         )
 

bright_vision_core/model_router_apply.py

@@ -4,7 +4,7 @@
 
 from cecli import models
 
-from bright_vision_core.model_router import ModelRouterConfig, RouteDecision
+from bright_vision_core.model_router import ModelRouterConfig, RouteDecision, normalize_keep_alive_for_tier
 
 
 def apply_route_to_coder(coder, decision: RouteDecision, router: ModelRouterConfig) -> None:
@@ -13,8 +13,9 @@ def apply_route_to_coder(coder, decision: RouteDecision, router: ModelRouterConf
     new_model = models.Model(decision.model_name, from_model=prev)
     if new_model.is_ollama():
         new_model._ensure_extra_params_dict()
-        keep_alive = (
-            router.keep_alive_fast if decision.tier == "fast" else router.keep_alive_heavy
+        keep_alive = normalize_keep_alive_for_tier(
+            decision.tier,
+            router.keep_alive_fast if decision.tier == "fast" else router.keep_alive_heavy,
         )
         new_model.extra_params["keep_alive"] = keep_alive
     coder.main_model = new_model