Add V2 worktree-bound claims

ARCHITECTURE.md

@@ -565,7 +565,7 @@ decide. No scheduler, no assignment, no orchestrator.
 **Backend capabilities manifest.** The `/health` endpoint returns `git_sha`
 and a `capabilities` list (`claims`, `structured_tasks`, `task_ids`,
 `checkpoint_notes`, `branch_disposition`, `freshness`, `compact_state`,
-`worktrees`).
+`worktrees`, `worktree_binding`).
 Agents probe this when a 404 or missing section suggests the backend is
 running stale code. The capabilities list is hardcoded in `main.py` and
 updated when new features ship.
@@ -573,10 +573,14 @@ updated when new features ship.
 **Worktree isolation.** Worktrees are the filesystem/index isolation primitive
 for multi-agent project work. `WorkTree` rows record an agent, absolute path,
 branch name, base git SHA, and lifecycle status while the backend creates and
-removes the corresponding git worktree on disk via `/api/v5/worktrees`. V1 is
+removes the corresponding git worktree on disk via `/api/v5/worktrees`.
-intentionally not claim-bound and does not appear in the state brief yet; a
+
-future integration pass should connect active claims to worktrees and surface
+V2 adds optional claim binding: `WorkClaim.worktree_id` can point at an active
-that operational state where agents already read `## Active work`.
+worktree, and the state endpoint enriches that active claim with cached git
+drift data (path, branch, dirty file count, ahead/behind vs `origin/main`, and
+last commit). This keeps worktree usage visible exactly where agents already
+look for coordination state — `## Active work` — while preserving solo-agent
+claims that do not need a worktree.
 
 ---
 

DECISIONS.md

@@ -475,6 +475,14 @@ are live coordination state (active, done, or abandoned with TTL expiry). Agents
 cross-reference at read time via the skill pack's selection logic, not via a
 stored link.
 
+### Why claim ↔ worktree binding is optional, not required
+
+Claims still need to work for solo agents, read-only investigations, and small
+non-overlapping work where opening a worktree would add ceremony without
+reducing risk. Making `worktree_id` optional lets Smriti surface filesystem
+isolation when contention risk exists while preserving the lightweight claim
+path for work that does not need a separate git index.
+
 ---
 
 ## Open questions and deferred decisions

REPO_STRUCTURE.md

@@ -54,6 +54,7 @@ smriti/
 │   │       ├── embedding.py    Embedding generation (pgvector)
 │   │       ├── parser.py       Transcript parsing utilities
 │   │       ├── pack_generator.py  Context pack rendering (V1 legacy)
+│   │       ├── worktree_probe.py  Cached git drift probe for bound claims
 │   │       └── llm/
 │   │           ├── base.py         LLM provider base class
 │   │           ├── mock_provider.py  Deterministic mock for testing
@@ -61,20 +62,22 @@ smriti/
 │   ├── config/
 │   │   ├── providers.example.yaml  Template — copy to providers.yaml
 │   │   └── providers.yaml          Your keys (gitignored, not committed)
-│   ├── alembic/                Database migrations (13 versions)
+│   ├── alembic/                Database migrations (14 versions)
 │   ├── tests/
-│   │   ├── integration/        API integration tests (116 tests)
+│   │   ├── integration/        API integration tests (122 tests)
 │   │   │   ├── test_api_v4_chat.py
 │   │   │   ├── test_api_v5_lineage.py
 │   │   │   ├── test_multi_branch_state.py
 │   │   │   ├── test_claims.py
+│   │   │   ├── test_claim_worktree_binding.py
 │   │   │   ├── test_worktrees.py
 │   │   │   ├── test_checkpoint_extract.py
 │   │   │   └── test_delete_endpoints.py
-│   │   └── unit/               Unit tests (120 tests)
+│   │   └── unit/               Unit tests (125 tests)
 │   │       ├── test_config_loader.py
 │   │       ├── test_extractor.py
 │   │       ├── test_golden_outputs.py
+│   │       ├── test_worktree_probe.py
 │   │       ├── test_worktree_paths.py
 │   │       ├── test_pack_generator.py
 │   │       └── test_parser.py
@@ -110,10 +113,10 @@ smriti/
 │   │   ├── formatters.py       Continuation-oriented markdown renderers
 │   │   │                         (multi-branch, active claims, divergence)
 │   │   └── skill_pack/         Agent skill pack source and renderer
-│   │       ├── template.md     Single source of truth (v1.9, 15 sections)
+│   │       ├── template.md     Single source of truth (v2.0, 15 sections)
 │   │       ├── renderer.py     Pure-function render + versioned install
 │   │       └── targets.py      Target configs (claude-code, codex)
-│   └── tests/                  CLI + MCP tests (117 tests)
+│   └── tests/                  CLI + MCP tests (122 tests)
 │       ├── test_branch_close.py
 │       ├── test_init.py
 │       ├── test_mcp_server.py
@@ -140,7 +143,7 @@ smriti/
 | `/api/v1` | `sessions.py` | Legacy | Transcript paste ingestion |
 | `/api/v2` | `repos.py`, `commits.py` | Current | Space CRUD, checkpoint read/list. `CommitResponse` includes `assumptions` and `artifacts`. |
 | `/api/v4` | `chat.py` | Current | Chat sessions, send_message, commit, head, multi-branch state (`/state` with active branches, active claims, and divergence signal). Provider status. |
-| `/api/v5` | `checkpoint.py`, `lineage.py`, `claims.py`, `worktrees.py` | Current | Checkpoint draft/review/extract, fork, lineage tree, compare, work claims, git worktrees. |
+| `/api/v5` | `checkpoint.py`, `lineage.py`, `claims.py`, `worktrees.py` | Current | Checkpoint draft/review/extract, fork, lineage tree, compare, work claims with optional worktree binding, git worktrees. |
 
 ---
 
@@ -161,11 +164,11 @@ make migration      Create a new migration (usage: make migration msg="...")
 
 ---
 
-## Test counts (as of V1 worktree primitive)
+## Test counts (as of V2 worktree binding)
 
 | Suite | Count | Location |
 |---|---|---|
-| Backend integration | 116 | `backend/tests/integration/` |
+| Backend integration | 122 | `backend/tests/integration/` |
-| Backend unit | 120 | `backend/tests/unit/` |
+| Backend unit | 125 | `backend/tests/unit/` |
-| CLI + MCP | 117 | `cli/tests/` |
+| CLI + MCP | 122 | `cli/tests/` |
-| **Total** | **353** | |
+| **Total** | **369** | |

backend/alembic/versions/2026_05_04_0200_c0db5e90f1a2_add_worktree_id_to_work_claims.py

@@ -0,0 +1,51 @@
+"""Add worktree_id FK to work_claims.
+
+Revision ID: c0db5e90f1a2
+Revises: b9cadbef0102
+Create Date: 2026-05-04 02:00:00.000000
+
+"""
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c0db5e90f1a2"
+down_revision = "b9cadbef0102"
+branch_labels = None
+depends_on = None
+
+WORKTREE_FK_NAME = "fk_work_claims_worktree_id_work_trees"
+WORKTREE_INDEX_NAME = "idx_work_claims_worktree"
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("work_claims") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "worktree_id",
+                postgresql.UUID(as_uuid=True),
+                nullable=True,
+            ),
+        )
+        batch_op.create_foreign_key(
+            WORKTREE_FK_NAME,
+            "work_trees",
+            ["worktree_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+    op.create_index(
+        WORKTREE_INDEX_NAME,
+        "work_claims",
+        ["worktree_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(WORKTREE_INDEX_NAME, table_name="work_claims")
+    with op.batch_alter_table("work_claims") as batch_op:
+        batch_op.drop_constraint(WORKTREE_FK_NAME, type_="foreignkey")
+        batch_op.drop_column("worktree_id")

backend/app/api/routes/chat.py

@@ -31,9 +31,17 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.db.database import get_db
-from app.db.models import ChatSession, CommitModel, RepoModel, TurnEvent
+from app.db.models import (
+    ChatSession,
+    CommitModel,
+    RepoModel,
+    TurnEvent,
+    WorkClaim,
+    WorkTree,
+)
 from app.config_loader import get_config, providers_status, ProviderNotConfiguredError
 from app.providers.registry import get_adapter, get_mock_adapter
+from app.services.worktree_probe import _probe_worktree
 
 router = APIRouter(prefix="/chat", tags=["chat-v4"])
 
@@ -358,6 +366,18 @@ class DivergenceSummary(BaseModel):
     pairs: list[DivergencePair] = Field(default_factory=list)
 
 
+class ActiveWorktreeSummary(BaseModel):
+    """Git drift information for a worktree bound to an active claim."""
+    id: uuid.UUID
+    path: str
+    branch: str
+    dirty_files: int
+    ahead: int
+    behind: int
+    last_commit_sha: str
+    last_commit_relative: str
+
+
 class ActiveClaimSummary(BaseModel):
     """One line per active work claim in the `Active work` section."""
     id: uuid.UUID
@@ -365,6 +385,8 @@ class ActiveClaimSummary(BaseModel):
     branch_name: str
     scope: str
     task_id: Optional[str] = None
+    worktree_id: Optional[uuid.UUID] = None
+    worktree: Optional[ActiveWorktreeSummary] = None
     intent_type: str
     claimed_at: datetime
     expires_at: datetime
@@ -993,7 +1015,6 @@ def get_space_state(
         divergence = _compute_space_divergence(main_head_commit, active_branch_commits)
 
     # Active work claims — query-time expiration filter.
-    from app.db.models import WorkClaim
     now = _utcnow()
     claims_stmt = (
         select(WorkClaim)
@@ -1011,6 +1032,17 @@ def get_space_state(
         if wc.base_commit_id:
             base_commit = db.get(CommitModel, wc.base_commit_id)
             base_hash = base_commit.commit_hash[:7] if base_commit else None
+        worktree_summary = None
+        if wc.worktree_id:
+            worktree = db.get(WorkTree, wc.worktree_id)
+            if worktree and worktree.status == "active":
+                probed = _probe_worktree(
+                    str(worktree.id),
+                    worktree.path,
+                    worktree.branch_name,
+                )
+                if probed:
+                    worktree_summary = ActiveWorktreeSummary(**probed)
         active_claims.append(
             ActiveClaimSummary(
                 id=wc.id,
@@ -1018,6 +1050,8 @@ def get_space_state(
                 branch_name=wc.branch_name,
                 scope=wc.scope,
                 task_id=wc.task_id,
+                worktree_id=wc.worktree_id,
+                worktree=worktree_summary,
                 intent_type=wc.intent_type,
                 claimed_at=wc.claimed_at,
                 expires_at=wc.expires_at,

backend/app/api/routes/claims.py

@@ -26,7 +26,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.db.database import get_db
-from app.db.models import CommitModel, RepoModel, WorkClaim
+from app.db.models import CommitModel, RepoModel, WorkClaim, WorkTree
 
 router = APIRouter(prefix="/claims", tags=["claims-v5"])
 
@@ -62,6 +62,10 @@ class CreateClaimRequest(BaseModel):
         default=None,
         description="Optional reference to a structured task's id from the checkpoint.",
     )
+    worktree_id: Optional[str] = Field(
+        default=None,
+        description="Optional worktree UUID this claim is bound to.",
+    )
     intent_type: str = "implement"
     ttl_hours: float = Field(
         default=DEFAULT_TTL_HOURS,
@@ -82,6 +86,7 @@ class ClaimResponse(BaseModel):
     branch_name: str
     base_commit_id: Optional[uuid.UUID] = None
     task_id: Optional[str] = None
+    worktree_id: Optional[uuid.UUID] = None
     scope: str
     intent_type: str
     status: str
@@ -132,6 +137,23 @@ def create_claim(payload: CreateClaimRequest, db: Session = Depends(get_db)):
         except ValueError:
             raise HTTPException(status_code=400, detail="Invalid session_id")
 
+    worktree_id = None
+    if payload.worktree_id:
+        try:
+            worktree_id = uuid.UUID(payload.worktree_id)
+        except ValueError:
+            raise HTTPException(status_code=400, detail="Invalid worktree_id")
+        worktree = db.get(WorkTree, worktree_id)
+        if not worktree:
+            raise HTTPException(status_code=404, detail="Worktree not found")
+        if worktree.repo_id != space_id:
+            raise HTTPException(
+                status_code=400,
+                detail="Worktree belongs to a different space",
+            )
+        if worktree.status != "active":
+            raise HTTPException(status_code=400, detail="Worktree is not active")
+
     now = _utcnow()
     claim = WorkClaim(
         repo_id=space_id,
@@ -139,6 +161,7 @@ def create_claim(payload: CreateClaimRequest, db: Session = Depends(get_db)):
         agent=payload.agent,
         branch_name=payload.branch_name,
         base_commit_id=base_commit_id,
+        worktree_id=worktree_id,
         scope=payload.scope,
         task_id=payload.task_id,
         intent_type=payload.intent_type,

backend/app/db/models.py

@@ -297,6 +297,10 @@ class WorkClaim(Base):
         UUID(as_uuid=True), ForeignKey("commits.id", ondelete="SET NULL"),
         nullable=True,
     )
+    worktree_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("work_trees.id", ondelete="SET NULL"),
+        nullable=True, index=True,
+    )
     scope: Mapped[str] = mapped_column(Text, nullable=False)
     task_id: Mapped[str | None] = mapped_column(
         String(100), nullable=True,
@@ -317,6 +321,8 @@ class WorkClaim(Base):
         DateTime(timezone=True), nullable=False,
     )
 
+    worktree: Mapped["WorkTree | None"] = relationship(back_populates="claims")
+
 
 class WorkTree(Base):
     """A git worktree allocated for an agent working in a space.
@@ -342,3 +348,4 @@ class WorkTree(Base):
     closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
     repo: Mapped["RepoModel"] = relationship(back_populates="worktrees")
+    claims: Mapped[list["WorkClaim"]] = relationship(back_populates="worktree")

backend/app/main.py

@@ -96,6 +96,7 @@ def create_app() -> FastAPI:
         "freshness",          # since_commit_id on state endpoint
         "compact_state",      # --compact mode on state brief
         "worktrees",          # /api/v5/worktrees
+        "worktree_binding",   # claims can bind to worktrees + state drift summary
     ]
 
     @app.get("/health")

backend/app/services/worktree_probe.py

@@ -0,0 +1,118 @@
+"""Cached git status probing for worktree-bound claims.
+
+The state endpoint uses this helper for active claims with a bound
+worktree. Probing must never make `smriti state` fail: stale, broken, or
+missing worktrees return None and the claim still renders normally.
+"""
+from __future__ import annotations
+
+import logging
+import subprocess
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+PROBE_TIMEOUT_SECONDS = 3
+PROBE_CACHE_TTL_SECONDS = 60
+_PROBE_CACHE: dict[str, tuple[float, dict[str, Any] | None]] = {}
+
+
+def clear_probe_cache() -> None:
+    """Clear the in-process probe cache. Used by tests."""
+    _PROBE_CACHE.clear()
+
+
+def _run_git(path: str, args: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", "-C", path, *args],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=PROBE_TIMEOUT_SECONDS,
+    )
+
+
+def _probe_worktree(worktree_id: str, path: str, branch: str) -> dict[str, Any] | None:
+    """Return git drift information for a worktree, or None on any error.
+
+    The cache is keyed by worktree_id so repeated `smriti state` calls do
+    not shell out on every request. Failures are cached too for the same
+    TTL to avoid repeatedly probing broken paths.
+    """
+    now = time.time()
+    cached = _PROBE_CACHE.get(worktree_id)
+    if cached and now - cached[0] < PROBE_CACHE_TTL_SECONDS:
+        return cached[1]
+
+    result = _probe_worktree_uncached(worktree_id, path, branch)
+    _PROBE_CACHE[worktree_id] = (now, result)
+    return result
+
+
+def _probe_worktree_uncached(
+    worktree_id: str,
+    path: str,
+    branch: str,
+) -> dict[str, Any] | None:
+    try:
+        status = _run_git(path, ["status", "--porcelain"])
+        if status.returncode != 0:
+            _log_probe_failure(worktree_id, "status", status)
+            return None
+        dirty_files = len([line for line in status.stdout.splitlines() if line.strip()])
+
+        counts = _run_git(path, ["rev-list", "--left-right", "--count", "HEAD...origin/main"])
+        if counts.returncode != 0:
+            _log_probe_failure(worktree_id, "rev-list", counts)
+            return None
+        parts = counts.stdout.strip().split()
+        if len(parts) != 2:
+            logger.warning(
+                "Malformed worktree ahead/behind output for %s: %r",
+                worktree_id,
+                counts.stdout,
+            )
+            return None
+        ahead, behind = int(parts[0]), int(parts[1])
+
+        last = _run_git(path, ["log", "-1", "--format=%h %ar"])
+        if last.returncode != 0:
+            _log_probe_failure(worktree_id, "log", last)
+            return None
+        last_parts = last.stdout.strip().split(maxsplit=1)
+        if len(last_parts) != 2:
+            logger.warning(
+                "Malformed worktree last-commit output for %s: %r",
+                worktree_id,
+                last.stdout,
+            )
+            return None
+
+        return {
+            "id": worktree_id,
+            "path": path,
+            "branch": branch,
+            "dirty_files": dirty_files,
+            "ahead": ahead,
+            "behind": behind,
+            "last_commit_sha": last_parts[0],
+            "last_commit_relative": last_parts[1],
+        }
+    except (FileNotFoundError, subprocess.TimeoutExpired, ValueError) as exc:
+        logger.warning("Worktree probe failed for %s: %s", worktree_id, exc)
+        return None
+
+
+def _log_probe_failure(
+    worktree_id: str,
+    command: str,
+    result: subprocess.CompletedProcess[str],
+) -> None:
+    detail = (result.stderr or result.stdout or "").strip()
+    logger.warning(
+        "Worktree probe command %s failed for %s: %s",
+        command,
+        worktree_id,
+        detail or f"exit {result.returncode}",
+    )

backend/tests/integration/test_api_v4_chat.py

@@ -2,7 +2,10 @@
 V4 Chat API integration tests.
 All tests use use_mock=True so they run without any real provider API keys.
 """
-import pytest
+import uuid
+
+from app.api.routes import chat
+from app.db.models import WorkTree
 
 
 def test_provider_status(client):
@@ -16,12 +19,91 @@ def test_provider_status(client):
         assert "has_key" in status
 
 
+def test_state_active_claims_include_bound_worktree_info(client, db_session, monkeypatch):
+    repo_id = _create_repo(client, "State Worktree Binding")
+    session_id = _create_session(client, repo_id)
+    _commit(client, repo_id, session_id, "base")
+    worktree = WorkTree(
+        repo_id=uuid.UUID(repo_id),
+        agent="codex-local",
+        path="/tmp/state-worktree",
+        branch_name="smriti/codex-local/abc12345",
+        base_commit_sha="abc123",
+        status="active",
+    )
+    db_session.add(worktree)
+    db_session.commit()
+    db_session.refresh(worktree)
+
+    claim_r = client.post(
+        "/api/v5/claims",
+        json={
+            "space_id": repo_id,
+            "agent": "codex-local",
+            "scope": "State enrichment test",
+            "branch_name": "worktree-v2-binding-and-enrichment",
+            "intent_type": "implement",
+            "worktree_id": str(worktree.id),
+        },
+    )
+    assert claim_r.status_code == 201, claim_r.text
+
+    def fake_probe(worktree_id, path, branch):
+        assert worktree_id == str(worktree.id)
+        assert path == "/tmp/state-worktree"
+        assert branch == "smriti/codex-local/abc12345"
+        return {
+            "id": worktree_id,
+            "path": path,
+            "branch": branch,
+            "dirty_files": 3,
+            "ahead": 1,
+            "behind": 0,
+            "last_commit_sha": "def5678",
+            "last_commit_relative": "5 minutes ago",
+        }
+
+    monkeypatch.setattr(chat, "_probe_worktree", fake_probe)
+
+    r = client.get(f"/api/v4/chat/spaces/{repo_id}/state")
+
+    assert r.status_code == 200, r.text
+    claim = r.json()["active_claims"][0]
+    assert claim["worktree_id"] == str(worktree.id)
+    assert claim["worktree"] == {
+        "id": str(worktree.id),
+        "path": "/tmp/state-worktree",
+        "branch": "smriti/codex-local/abc12345",
+        "dirty_files": 3,
+        "ahead": 1,
+        "behind": 0,
+        "last_commit_sha": "def5678",
+        "last_commit_relative": "5 minutes ago",
+    }
+
+
 def _create_repo(client, name="Chat Test Repo"):
     r = client.post("/api/v2/repos", json={"name": name})
     assert r.status_code == 201, r.text
     return r.json()["id"]
 
 
+def _create_session(client, repo_id, title="test"):
+    r = client.post(
+        f"/api/v4/chat/spaces/{repo_id}/sessions",
+        json={"title": title, "provider": "openrouter", "model": "mock"},
+    )
+    assert r.status_code == 201, r.text
+    return r.json()["id"]
+
+
+def _commit(client, repo_id, session_id, message="checkpoint", **kwargs):
+    payload = {"repo_id": repo_id, "session_id": session_id, "message": message, **kwargs}
+    r = client.post("/api/v4/chat/commit", json=payload)
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
 def test_session_lifecycle(client):
     repo_id = _create_repo(client)
 

backend/tests/integration/test_claim_worktree_binding.py

@@ -0,0 +1,92 @@
+"""Integration tests for claim to worktree binding."""
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+
+from app.db.models import WorkTree
+
+
+def _create_repo(client, name="Claim Worktree Test Repo"):
+    r = client.post("/api/v2/repos", json={"name": name})
+    assert r.status_code == 201, r.text
+    return r.json()["id"]
+
+
+def _create_worktree_row(db_session, repo_id: str, *, status="active") -> WorkTree:
+    worktree = WorkTree(
+        repo_id=uuid.UUID(repo_id),
+        agent="codex-local",
+        path=f"/tmp/wt-{uuid.uuid4().hex[:8]}",
+        branch_name=f"smriti/codex-local/{uuid.uuid4().hex[:8]}",
+        base_commit_sha="abc123",
+        status=status,
+        created_at=datetime.now(UTC),
+    )
+    db_session.add(worktree)
+    db_session.commit()
+    db_session.refresh(worktree)
+    return worktree
+
+
+def _create_claim(client, space_id: str, **overrides):
+    payload = {
+        "space_id": space_id,
+        "agent": "codex-local",
+        "scope": "Bound worktree claim",
+        "branch_name": "worktree-v2-binding-and-enrichment",
+        "intent_type": "implement",
+        "ttl_hours": 4.0,
+        **overrides,
+    }
+    return client.post("/api/v5/claims", json=payload)
+
+
+def test_create_claim_with_worktree_id_links_records(client, db_session):
+    repo_id = _create_repo(client)
+    worktree = _create_worktree_row(db_session, repo_id)
+
+    r = _create_claim(client, repo_id, worktree_id=str(worktree.id))
+
+    assert r.status_code == 201, r.text
+    claim = r.json()
+    assert claim["worktree_id"] == str(worktree.id)
+
+
+def test_create_claim_with_nonexistent_worktree_id_returns_404(client):
+    repo_id = _create_repo(client)
+
+    r = _create_claim(client, repo_id, worktree_id=str(uuid.uuid4()))
+
+    assert r.status_code == 404
+    assert "Worktree not found" in r.json()["detail"]
+
+
+def test_create_claim_with_worktree_from_other_space_returns_400(client, db_session):
+    repo_id = _create_repo(client, "Claim Space")
+    other_repo_id = _create_repo(client, "Other Space")
+    other_worktree = _create_worktree_row(db_session, other_repo_id)
+
+    r = _create_claim(client, repo_id, worktree_id=str(other_worktree.id))
+
+    assert r.status_code == 400
+    assert "different space" in r.json()["detail"]
+
+
+def test_create_claim_with_closed_worktree_returns_400(client, db_session):
+    repo_id = _create_repo(client)
+    worktree = _create_worktree_row(db_session, repo_id, status="closed")
+
+    r = _create_claim(client, repo_id, worktree_id=str(worktree.id))
+
+    assert r.status_code == 400
+    assert "not active" in r.json()["detail"]
+
+
+def test_create_claim_without_worktree_id_preserves_existing_behavior(client):
+    repo_id = _create_repo(client)
+
+    r = _create_claim(client, repo_id)
+
+    assert r.status_code == 201, r.text
+    assert r.json()["worktree_id"] is None

backend/tests/integration/test_health.py

@@ -35,6 +35,7 @@ def test_health_includes_required_capabilities(client):
         "freshness",
         "compact_state",
         "worktrees",
+        "worktree_binding",
     ]
     for cap in required:
         assert cap in caps, f"Missing capability: {cap}"

backend/tests/unit/test_worktree_probe.py

@@ -0,0 +1,110 @@
+"""Unit tests for cached worktree git probing."""
+from __future__ import annotations
+
+import subprocess
+
+from app.services import worktree_probe
+
+
+def _result(returncode=0, stdout="", stderr=""):
+    return subprocess.CompletedProcess(
+        args=["git"],
+        returncode=returncode,
+        stdout=stdout,
+        stderr=stderr,
+    )
+
+
+def test_probe_worktree_success(monkeypatch):
+    worktree_probe.clear_probe_cache()
+    calls: list[list[str]] = []
+
+    def fake_run_git(path, args):
+        calls.append(args)
+        if args == ["status", "--porcelain"]:
+            return _result(stdout=" M file.py\n?? new.py\n")
+        if args == ["rev-list", "--left-right", "--count", "HEAD...origin/main"]:
+            return _result(stdout="2\t1\n")
+        if args == ["log", "-1", "--format=%h %ar"]:
+            return _result(stdout="abc1234 5 minutes ago\n")
+        raise AssertionError(args)
+
+    monkeypatch.setattr(worktree_probe, "_run_git", fake_run_git)
+
+    result = worktree_probe._probe_worktree("wt-1", "/tmp/wt", "feature/wt")
+
+    assert result == {
+        "id": "wt-1",
+        "path": "/tmp/wt",
+        "branch": "feature/wt",
+        "dirty_files": 2,
+        "ahead": 2,
+        "behind": 1,
+        "last_commit_sha": "abc1234",
+        "last_commit_relative": "5 minutes ago",
+    }
+    assert len(calls) == 3
+
+
+def test_probe_worktree_cache_hit(monkeypatch):
+    worktree_probe.clear_probe_cache()
+    calls = 0
+
+    def fake_run_git(path, args):
+        nonlocal calls
+        calls += 1
+        if args == ["status", "--porcelain"]:
+            return _result(stdout="")
+        if args == ["rev-list", "--left-right", "--count", "HEAD...origin/main"]:
+            return _result(stdout="0\t0\n")
+        return _result(stdout="abc1234 just now\n")
+
+    monkeypatch.setattr(worktree_probe, "_run_git", fake_run_git)
+
+    first = worktree_probe._probe_worktree("wt-cache", "/tmp/wt", "branch")
+    second = worktree_probe._probe_worktree("wt-cache", "/tmp/wt", "branch")
+
+    assert first == second
+    assert calls == 3
+
+
+def test_probe_worktree_timeout_returns_none_and_caches(monkeypatch):
+    worktree_probe.clear_probe_cache()
+    calls = 0
+
+    def fake_run_git(path, args):
+        nonlocal calls
+        calls += 1
+        raise subprocess.TimeoutExpired(["git"], timeout=3)
+
+    monkeypatch.setattr(worktree_probe, "_run_git", fake_run_git)
+
+    assert worktree_probe._probe_worktree("wt-timeout", "/tmp/wt", "branch") is None
+    assert worktree_probe._probe_worktree("wt-timeout", "/tmp/wt", "branch") is None
+    assert calls == 1
+
+
+def test_probe_worktree_nonzero_exit_returns_none(monkeypatch):
+    worktree_probe.clear_probe_cache()
+    monkeypatch.setattr(
+        worktree_probe,
+        "_run_git",
+        lambda path, args: _result(returncode=128, stderr="fatal"),
+    )
+
+    assert worktree_probe._probe_worktree("wt-fail", "/tmp/wt", "branch") is None
+
+
+def test_probe_worktree_malformed_output_returns_none(monkeypatch):
+    worktree_probe.clear_probe_cache()
+
+    def fake_run_git(path, args):
+        if args == ["status", "--porcelain"]:
+            return _result(stdout="")
+        if args == ["rev-list", "--left-right", "--count", "HEAD...origin/main"]:
+            return _result(stdout="not-two-fields\n")
+        raise AssertionError(args)
+
+    monkeypatch.setattr(worktree_probe, "_run_git", fake_run_git)
+
+    assert worktree_probe._probe_worktree("wt-bad", "/tmp/wt", "branch") is None

cli/smriti_cli/client.py

@@ -181,6 +181,7 @@ class SmritiClient:
         branch_name: str = "main",
         base_commit_id: str | None = None,
         task_id: str | None = None,
+        worktree_id: str | None = None,
         intent_type: str = "implement",
         ttl_hours: float = 4.0,
     ) -> dict:
@@ -197,6 +198,8 @@ class SmritiClient:
             payload["base_commit_id"] = base_commit_id
         if task_id:
             payload["task_id"] = task_id
+        if worktree_id:
+            payload["worktree_id"] = worktree_id
         return self._request("POST", "/api/v5/claims", json=payload)
 
     def update_claim(self, claim_id: str, status: str) -> dict:

cli/smriti_cli/formatters.py

@@ -260,6 +260,23 @@ def _format_active_claims_section(active_claims: list[dict]) -> str:
             f"- `{agent}` [{intent}] on `{branch}` from `{base_hash}` "
             f"· {created} — {scope}{task_suffix}"
         )
+        worktree_id = c.get("worktree_id")
+        worktree = c.get("worktree")
+        if worktree:
+            path = _pretty_path(worktree.get("path")) or worktree.get("path") or "?"
+            dirty = worktree.get("dirty_files", 0)
+            ahead = worktree.get("ahead", 0)
+            behind = worktree.get("behind", 0)
+            last_sha = worktree.get("last_commit_sha") or "?"
+            last_rel = worktree.get("last_commit_relative") or "?"
+            lines.append(f"   · worktree: {path}")
+            lines.append(
+                f"   · branch: {worktree.get('branch') or '?'} · "
+                f"{dirty} dirty · ahead {ahead} · behind {behind} · "
+                f"last commit `{last_sha}` {last_rel}"
+            )
+        elif worktree_id:
+            lines.append("   · worktree: (probe failed or worktree closed)")
     return "\n".join(lines) + "\n"
 
 

cli/smriti_cli/main.py

@@ -758,6 +758,7 @@ def cmd_claim_create(client: SmritiClient, args: argparse.Namespace) -> None:
         branch_name=args.branch or "main",
         base_commit_id=base_commit_id,
         task_id=getattr(args, "task_id", None),
+        worktree_id=getattr(args, "worktree", None),
         intent_type=args.intent_type,
         ttl_hours=args.ttl,
     )
@@ -1224,6 +1225,7 @@ def _build_parser() -> argparse.ArgumentParser:
         help="Type of work (default: implement)",
     )
     cl_create.add_argument("--task-id", dest="task_id", default=None, help="Optional structured task ID this claim covers")
+    cl_create.add_argument("--worktree", dest="worktree", default=None, help="Optional worktree UUID to bind to this claim")
     cl_create.add_argument("--ttl", type=float, default=4.0, help="Hours until expiration (default: 4)")
     cl_create.add_argument("--json", action="store_true")
     cl_create.set_defaults(func=cmd_claim_create)

cli/smriti_cli/mcp_server.py

@@ -545,6 +545,7 @@ def smriti_claim(
     agent: str = "claude-code",
     branch: str = "main",
     task_id: str = "",
+    worktree_id: str = "",
     intent_type: str = "implement",
     ttl_hours: float = 4.0,
 ) -> str:
@@ -566,6 +567,7 @@ def smriti_claim(
         agent: Your agent identifier (default: claude-code).
         branch: Branch you will work on (default: main).
         task_id: Optional ID of the structured task this claim covers (from the checkpoint's task list).
+        worktree_id: Optional worktree UUID to bind to this claim.
         intent_type: One of: implement, review, investigate, docs, test.
         ttl_hours: Hours until the claim expires (default: 4).
     """
@@ -581,6 +583,7 @@ def smriti_claim(
             branch_name=branch,
             base_commit_id=base_commit_id,
             task_id=task_id or None,
+            worktree_id=worktree_id or None,
             intent_type=intent_type,
             ttl_hours=ttl_hours,
         )

cli/smriti_cli/skill_pack/template.md

@@ -1,5 +1,5 @@
 ---
-smriti_skill_pack_version: 1.9
+smriti_skill_pack_version: 2.0
 title: Smriti — how to use it well
 target: {{display_name}}
 ---
@@ -241,6 +241,8 @@ are a client of it. You do not own it.
   at localhost:8000 does not support [feature]. Its git_sha is
   [sha] but the current repo is at [repo sha]. Please restart
   the backend with `make dev` to pick up recent changes."
+  For worktree-aware coordination, the capabilities list should include
+  both `worktrees` and `worktree_binding`.
 - **When to check capabilities:** You do NOT need to check on every
   session. Check when:
   - A Smriti API call returns 404 on a route you expect to exist
@@ -307,6 +309,65 @@ so other agents can see your base state.
 Say out loud: **"Claiming: [intent_type] — <scope>."** before
 creating the claim.
 
+### 3.6.1 Worktrees: when and how
+
+If multiple agents are working on the same project on the same machine,
+each agent should work in its own git worktree. Sharing a working tree
+across agents is the single highest-cost failure mode — staged files
+from one agent can land in another agent's commit, and the wrong code
+ships to main. The chaanbeen-web retros documented exactly this incident.
+
+The reflex: when you start substantial work on a project where another
+agent might be active, open a worktree before your first edit.
+
+```
+{{mcp:smriti_worktree_open(space="<project>", agent="<your-id>")}}{{cli:smriti worktree open <project> --agent <your-id>}}
+```
+
+This returns a path. Use that path as your working directory for the
+rest of the session. Your edits, your staging index, your commits all
+live in that worktree. Other agents have their own worktrees; their
+filesystem state is invisible to you and yours to them.
+
+When you create your work claim, bind it to the worktree:
+
+```
+{{mcp:smriti_claim(space="<project>", scope="...", agent="<your-id>", intent_type="implement", task_id="<task-id>", worktree_id="<worktree-id>")}}{{cli:smriti claim create <project> --agent <your-id> \
+    --scope "..." --intent-type implement \
+    --task-id <task-id> --worktree <worktree-id>}}
+```
+
+MCP tools call this field `worktree_id`; the CLI flag is `--worktree`.
+
+The state brief now shows a worktree info line under your claim:
+which path, which branch, how many dirty files, ahead/behind vs main,
+last commit. Other agents seeing your claim know exactly what state
+your tree is in without asking.
+
+When your work is done and merged, close the worktree:
+
+```
+{{mcp:smriti_worktree_close(worktree_id="<worktree-id>")}}{{cli:smriti worktree close <worktree-id>}}
+```
+
+This refuses if you have uncommitted changes (correct default — stop
+and decide before destroying work). Pass `--force` only after you've
+confirmed the dirty changes are intentionally being discarded.
+
+When NOT to open a worktree:
+
+- Solo work on a project with no other active agents.
+- Quick read-only investigations that won't produce commits.
+- Documentation-only work that's clearly disjoint from anyone else's
+  track (e.g. you're writing in `docs/` while another agent is in
+  `backend/`). Worktrees are cheap but not zero cost; the discipline
+  is "open one when there's actual filesystem contention risk,"
+  not "open one for every session."
+
+Say out loud: **"Opening a worktree for this session."** before the
+first worktree open. **"Binding my claim to worktree <id>."** when
+claiming. **"Closing the worktree."** when done.
+
 ### 3.7 Check freshness before checkpointing
 
 If you have been working for more than a few minutes, check whether
@@ -773,6 +834,12 @@ Section 5.
   Duplicating completed work wastes a full session and creates noise
   in the timeline.
 
+- **Do not share a working tree between agents on the same project on
+  the same machine.** Open a worktree per agent. The chaanbeen
+  retrospectives documented one cross-agent commit pollution incident
+  that took ~1 hour to recover and degraded prod for ~6 endpoints —
+  this is the failure worktrees were built to prevent.
+
 - **Do not use `smriti_install_skill` to overwrite an in-project
   skill pack that you did not write.** If the project already has
   a skill pack of an older version, the install tool will tell you.
@@ -796,6 +863,9 @@ Use them literally.
 | reconcile state against repo | "Checking whether the flagged tasks are already reflected in the repo before starting." |
 | verify repo hygiene at start | "Repo is clean and synced against origin." or "Found local residue — classifying before proceeding." |
 | declare a work claim | "Claiming: [intent_type] — <scope>." |
+| open a worktree | "Opening a worktree for this session." |
+| bind a claim to a worktree | "Binding my claim to worktree <id>." |
+| close a worktree | "Closing the worktree." |
 | add a note to a checkpoint | "Adding a [kind] note to checkpoint X." |
 | self-select complementary work | "Selecting complementary work from the task list." |
 | check freshness before checkpoint | "Checking freshness before checkpointing." |
@@ -856,7 +926,7 @@ tell you. Do not guess.
 
 ---
 
-*Smriti skill pack version {{primary_mode}}-1.9 — this file is
+*Smriti skill pack version {{primary_mode}}-2.0 — this file is
 authoritative for agent behaviour on this project. If you catch it
 contradicting itself or your observed behaviour of the tools, tell
 the human; the skill pack is versioned and meant to be updated.*

cli/tests/test_skill_pack.py

@@ -44,7 +44,7 @@ def test_load_template_nonempty():
 
 def test_get_version_parses_frontmatter():
     version = get_version()
-    assert version == "1.9"
+    assert version == "2.0"
 
 
 def test_get_version_raises_when_frontmatter_missing():
@@ -148,6 +148,7 @@ _REQUIRED_PHRASES = [
     "git_sha",
     "stale code",
     "do not attempt to start",
+    "worktree_binding",
     # Section 3.6 — work claims
     "work claims",
     "declare intent",
@@ -155,6 +156,14 @@ _REQUIRED_PHRASES = [
     "not a lock",
     "not a scheduler",
     "leave active work claims hanging",
+    # Section 3.6.1 — worktree reflex
+    "Worktrees: when and how",
+    "Opening a worktree for this session",
+    "Binding my claim to worktree",
+    "Closing the worktree",
+    "worktree_id",
+    "cross-agent commit pollution",
+    "Open a worktree per agent",
     # Section 3.7 — freshness check
     "check freshness",
     "since",

cli/tests/test_state_multi_branch.py

@@ -652,3 +652,60 @@ def test_claims_section_without_task_id():
     out = _format_active_claims_section(claims)
     assert "(task:" not in out
     assert "[implement]" in out
+
+
+def test_claims_section_with_worktree_info():
+    """Bound worktree info renders as continuation lines under the claim."""
+    from smriti_cli.formatters import _format_active_claims_section
+    claims = [
+        {
+            "agent": "codex-local",
+            "branch_name": "worktree-v2-binding-and-enrichment",
+            "scope": "Implement V2",
+            "task_id": "v2-plan",
+            "worktree_id": "worktree-uuid",
+            "worktree": {
+                "id": "worktree-uuid",
+                "path": "/Users/example/.smriti/worktrees/smriti-dev/codex-local-abc12345",
+                "branch": "smriti/codex-local/abc12345",
+                "dirty_files": 3,
+                "ahead": 1,
+                "behind": 0,
+                "last_commit_sha": "def5678",
+                "last_commit_relative": "5 minutes ago",
+            },
+            "intent_type": "implement",
+            "base_commit_hash": "abc1234",
+            "claimed_at": datetime.now(timezone.utc).isoformat(),
+        },
+    ]
+
+    out = _format_active_claims_section(claims)
+
+    assert "worktree:" in out
+    assert "smriti/codex-local/abc12345" in out
+    assert "3 dirty" in out
+    assert "ahead 1" in out
+    assert "behind 0" in out
+    assert "last commit `def5678` 5 minutes ago" in out
+
+
+def test_claims_section_with_worktree_probe_failure():
+    """A bound claim with failed probing still renders a useful hint."""
+    from smriti_cli.formatters import _format_active_claims_section
+    claims = [
+        {
+            "agent": "codex-local",
+            "branch_name": "main",
+            "scope": "Implement V2",
+            "worktree_id": "worktree-uuid",
+            "worktree": None,
+            "intent_type": "implement",
+            "base_commit_hash": "abc1234",
+            "claimed_at": datetime.now(timezone.utc).isoformat(),
+        },
+    ]
+
+    out = _format_active_claims_section(claims)
+
+    assert "probe failed or worktree closed" in out

cli/tests/test_worktree_cli.py

@@ -148,3 +148,61 @@ def test_cmd_worktree_close_calls_client(capsys: pytest.CaptureFixture[str]):
     out = capsys.readouterr().out
     assert "Closed worktree" in out
     client.close_worktree.assert_called_once_with("wt-uuid", force=True)
+
+
+def test_claim_create_parser_accepts_worktree_flag():
+    parser = cli_main._build_parser()
+
+    args = parser.parse_args(
+        [
+            "claim",
+            "create",
+            "my-project",
+            "--agent",
+            "codex-local",
+            "--scope",
+            "Implement V2",
+            "--worktree",
+            "wt-uuid",
+        ]
+    )
+
+    assert args.worktree == "wt-uuid"
+
+
+def test_cmd_claim_create_passes_worktree_id():
+    client = MagicMock(spec=SmritiClient)
+    client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
+    client.get_head.return_value = {"commit_id": "checkpoint-uuid"}
+    client.create_claim.return_value = {
+        "id": "claim-uuid",
+        "agent": "codex-local",
+        "scope": "Implement V2",
+        "branch_name": "worktree-v2-binding-and-enrichment",
+        "intent_type": "implement",
+    }
+    args = argparse.Namespace(
+        space="my-project",
+        agent="codex-local",
+        scope="Implement V2",
+        branch="worktree-v2-binding-and-enrichment",
+        task_id="v2-plan",
+        worktree="wt-uuid",
+        intent_type="implement",
+        ttl=4.0,
+        json=False,
+    )
+
+    cli_main.cmd_claim_create(client, args)
+
+    client.create_claim.assert_called_once_with(
+        space_id="space-uuid",
+        agent="codex-local",
+        scope="Implement V2",
+        branch_name="worktree-v2-binding-and-enrichment",
+        base_commit_id="checkpoint-uuid",
+        task_id="v2-plan",
+        worktree_id="wt-uuid",
+        intent_type="implement",
+        ttl_hours=4.0,
+    )

cli/tests/test_worktree_mcp.py

@@ -75,3 +75,37 @@ def test_mcp_worktree_close_calls_client(mock_client):
 
     assert "Closed worktree" in result
     mock_client.close_worktree.assert_called_once_with("wt-uuid", force=True)
+
+
+def test_mcp_claim_passes_worktree_id(mock_client):
+    mock_client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
+    mock_client.get_head.return_value = {"commit_id": "checkpoint-uuid"}
+    mock_client.create_claim.return_value = {
+        "id": "claim-uuid",
+        "agent": "codex-local",
+        "scope": "Implement V2",
+        "branch_name": "worktree-v2-binding-and-enrichment",
+        "intent_type": "implement",
+    }
+
+    result = mcp_server.smriti_claim(
+        space="my-project",
+        scope="Implement V2",
+        agent="codex-local",
+        branch="worktree-v2-binding-and-enrichment",
+        task_id="v2-plan",
+        worktree_id="wt-uuid",
+    )
+
+    assert "Claimed" in result
+    mock_client.create_claim.assert_called_once_with(
+        space_id="space-uuid",
+        agent="codex-local",
+        scope="Implement V2",
+        branch_name="worktree-v2-binding-and-enrichment",
+        base_commit_id="checkpoint-uuid",
+        task_id="v2-plan",
+        worktree_id="wt-uuid",
+        intent_type="implement",
+        ttl_hours=4.0,
+    )