Add V1 worktree primitive

Adds WorkTree schema and migration, /api/v5/worktrees CRUD, CLI and MCP worktree surfaces, targeted regression tests, health capability, and minimal docs. Live Postgres migration/manual localhost verification intentionally remain pending until the full backend provider-config gate is resolved.
This commit is contained in:
Himanshu Dongre 2026-05-04 00:58:30 +05:30
parent 4625b5138a
commit b5ccdce758
14 changed files with 1291 additions and 19 deletions

View file

@ -349,7 +349,7 @@ Surfaces:
own reasoning in their own context, using their own LLM provider. Smriti is own reasoning in their own context, using their own LLM provider. Smriti is
their shared memory, not their runtime. their shared memory, not their runtime.
- **MCP server** (`cli/smriti_cli/mcp_server.py`, entry point `smriti-mcp`) — - **MCP server** (`cli/smriti_cli/mcp_server.py`, entry point `smriti-mcp`) —
the same surface wrapped as 17 MCP tools for hosts that speak the Model the same surface wrapped as 21 MCP tools for hosts that speak the Model
Context Protocol natively (Claude Code, Cursor, Windsurf). Stdio transport. Context Protocol natively (Claude Code, Cursor, Windsurf). Stdio transport.
Each tool is a thin shim: build a `SmritiClient`, call one or two methods, Each tool is a thin shim: build a `SmritiClient`, call one or two methods,
run the result through an existing formatter, return markdown. Feature run the result through an existing formatter, return markdown. Feature
@ -564,11 +564,20 @@ decide. No scheduler, no assignment, no orchestrator.
**Backend capabilities manifest.** The `/health` endpoint returns `git_sha` **Backend capabilities manifest.** The `/health` endpoint returns `git_sha`
and a `capabilities` list (`claims`, `structured_tasks`, `task_ids`, and a `capabilities` list (`claims`, `structured_tasks`, `task_ids`,
`checkpoint_notes`, `branch_disposition`, `freshness`, `compact_state`). `checkpoint_notes`, `branch_disposition`, `freshness`, `compact_state`,
`worktrees`).
Agents probe this when a 404 or missing section suggests the backend is 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 running stale code. The capabilities list is hardcoded in `main.py` and
updated when new features ship. 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
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
that operational state where agents already read `## Active work`.
--- ---
## What Is Not Yet In the Architecture ## What Is Not Yet In the Architecture

View file

@ -28,7 +28,7 @@ smriti/
│ │ ├── db/ │ │ ├── db/
│ │ │ ├── database.py SQLAlchemy engine and session factory │ │ │ ├── database.py SQLAlchemy engine and session factory
│ │ │ └── models.py ORM models: RepoModel, CommitModel, ChatSession, │ │ │ └── models.py ORM models: RepoModel, CommitModel, ChatSession,
│ │ │ TurnEvent, WorkClaim │ │ │ TurnEvent, WorkClaim, WorkTree
│ │ ├── domain/ │ │ ├── domain/
│ │ │ └── enums.py SessionStatus, TargetTool, etc. │ │ │ └── enums.py SessionStatus, TargetTool, etc.
│ │ ├── api/ │ │ ├── api/
@ -39,6 +39,7 @@ smriti/
│ │ │ ├── lineage.py V5: fork, branch tree, checkpoint compare, │ │ │ ├── lineage.py V5: fork, branch tree, checkpoint compare,
│ │ │ │ reachable checkpoints │ │ │ │ reachable checkpoints
│ │ │ ├── claims.py V5: work claims (create, update, list) │ │ │ ├── claims.py V5: work claims (create, update, list)
│ │ │ ├── worktrees.py V5: git worktrees (open, list, show, close)
│ │ │ ├── repos.py V2: Space CRUD, Checkpoint CRUD │ │ │ ├── repos.py V2: Space CRUD, Checkpoint CRUD
│ │ │ ├── commits.py V2: direct commit creation │ │ │ ├── commits.py V2: direct commit creation
│ │ │ └── context_git.py V2: context extraction │ │ │ └── context_git.py V2: context extraction
@ -60,19 +61,21 @@ smriti/
│ ├── config/ │ ├── config/
│ │ ├── providers.example.yaml Template — copy to providers.yaml │ │ ├── providers.example.yaml Template — copy to providers.yaml
│ │ └── providers.yaml Your keys (gitignored, not committed) │ │ └── providers.yaml Your keys (gitignored, not committed)
│ ├── alembic/ Database migrations (10 versions) │ ├── alembic/ Database migrations (13 versions)
│ ├── tests/ │ ├── tests/
│ │ ├── integration/ API integration tests (177 tests) │ │ ├── integration/ API integration tests (116 tests)
│ │ │ ├── test_api_v4_chat.py │ │ │ ├── test_api_v4_chat.py
│ │ │ ├── test_api_v5_lineage.py │ │ │ ├── test_api_v5_lineage.py
│ │ │ ├── test_multi_branch_state.py │ │ │ ├── test_multi_branch_state.py
│ │ │ ├── test_claims.py │ │ │ ├── test_claims.py
│ │ │ ├── test_worktrees.py
│ │ │ ├── test_checkpoint_extract.py │ │ │ ├── test_checkpoint_extract.py
│ │ │ └── test_delete_endpoints.py │ │ │ └── test_delete_endpoints.py
│ │ └── unit/ Unit tests (97 tests) │ │ └── unit/ Unit tests (120 tests)
│ │ ├── test_config_loader.py │ │ ├── test_config_loader.py
│ │ ├── test_extractor.py │ │ ├── test_extractor.py
│ │ ├── test_golden_outputs.py │ │ ├── test_golden_outputs.py
│ │ ├── test_worktree_paths.py
│ │ ├── test_pack_generator.py │ │ ├── test_pack_generator.py
│ │ └── test_parser.py │ │ └── test_parser.py
│ └── pyproject.toml Python dependencies (includes python-dotenv) │ └── pyproject.toml Python dependencies (includes python-dotenv)
@ -101,22 +104,24 @@ smriti/
│ ├── smriti_cli/ │ ├── smriti_cli/
│ │ ├── main.py argparse dispatcher: init, space, state, │ │ ├── main.py argparse dispatcher: init, space, state,
│ │ │ checkpoint, fork, restore, compare, │ │ │ checkpoint, fork, restore, compare,
│ │ │ branch, claim, skills │ │ │ branch, claim, worktree, skills
│ │ ├── mcp_server.py FastMCP server (17 tools, stdio transport) │ │ ├── mcp_server.py FastMCP server (21 tools, stdio transport)
│ │ ├── client.py SmritiClient HTTP wrapper (includes claims) │ │ ├── client.py SmritiClient HTTP wrapper (includes claims/worktrees)
│ │ ├── formatters.py Continuation-oriented markdown renderers │ │ ├── formatters.py Continuation-oriented markdown renderers
│ │ │ (multi-branch, active claims, divergence) │ │ │ (multi-branch, active claims, divergence)
│ │ └── skill_pack/ Agent skill pack source and renderer │ │ └── skill_pack/ Agent skill pack source and renderer
│ │ ├── template.md Single source of truth (v1.9, 15 sections) │ │ ├── template.md Single source of truth (v1.9, 15 sections)
│ │ ├── renderer.py Pure-function render + versioned install │ │ ├── renderer.py Pure-function render + versioned install
│ │ └── targets.py Target configs (claude-code, codex) │ │ └── targets.py Target configs (claude-code, codex)
│ └── tests/ CLI + MCP tests (107 tests) │ └── tests/ CLI + MCP tests (117 tests)
│ ├── test_branch_close.py │ ├── test_branch_close.py
│ ├── test_init.py │ ├── test_init.py
│ ├── test_mcp_server.py │ ├── test_mcp_server.py
│ ├── test_skill_pack.py │ ├── test_skill_pack.py
│ ├── test_smoke.py │ ├── test_smoke.py
│ └── test_state_multi_branch.py │ ├── test_state_multi_branch.py
│ ├── test_worktree_cli.py
│ └── test_worktree_mcp.py
├── docs/ ├── docs/
│ └── API.md V2, V4, and V5 endpoint reference │ └── API.md V2, V4, and V5 endpoint reference
@ -135,7 +140,7 @@ smriti/
| `/api/v1` | `sessions.py` | Legacy | Transcript paste ingestion | | `/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/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/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` | Current | Checkpoint draft/review/extract, fork, lineage tree, compare, work claims. | | `/api/v5` | `checkpoint.py`, `lineage.py`, `claims.py`, `worktrees.py` | Current | Checkpoint draft/review/extract, fork, lineage tree, compare, work claims, git worktrees. |
--- ---
@ -156,11 +161,11 @@ make migration Create a new migration (usage: make migration msg="...")
--- ---
## Test counts (as of v1.9 skill pack + task IDs + metrics) ## Test counts (as of V1 worktree primitive)
| Suite | Count | Location | | Suite | Count | Location |
|---|---|---| |---|---|---|
| Backend integration | 107 | `backend/tests/integration/` | | Backend integration | 116 | `backend/tests/integration/` |
| Backend unit | 117 | `backend/tests/unit/` | | Backend unit | 120 | `backend/tests/unit/` |
| CLI + MCP | 107 | `cli/tests/` | | CLI + MCP | 117 | `cli/tests/` |
| **Total** | **331** | | | **Total** | **353** | |

View file

@ -0,0 +1,61 @@
"""Add work_trees table.
Revision ID: b9cadbef0102
Revises: a8b9cadbef01
Create Date: 2026-05-04 01:00:00.000000
"""
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
# revision identifiers, used by Alembic.
revision = "b9cadbef0102"
down_revision = "a8b9cadbef01"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"work_trees",
sa.Column(
"id",
postgresql.UUID(as_uuid=True),
primary_key=True,
server_default=sa.text("gen_random_uuid()"),
),
sa.Column(
"repo_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("repos.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("agent", sa.String(100), nullable=False),
sa.Column("path", sa.Text(), nullable=False),
sa.Column("branch_name", sa.String(255), nullable=False),
sa.Column("base_commit_sha", sa.String(64), nullable=True),
sa.Column("status", sa.String(20), nullable=False, server_default="active"),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_index("idx_work_trees_repo", "work_trees", ["repo_id"], unique=False)
op.create_index(
"idx_work_trees_active",
"work_trees",
["repo_id", "status"],
unique=False,
postgresql_where=sa.text("status = 'active'"),
)
def downgrade() -> None:
op.drop_index("idx_work_trees_active", table_name="work_trees")
op.drop_index("idx_work_trees_repo", table_name="work_trees")
op.drop_table("work_trees")

View file

@ -0,0 +1,328 @@
"""
V5 Worktree API routes.
Worktrees provide filesystem-level isolation for agents: each agent gets
its own working directory and git index while sharing the same repository
object store. V1 intentionally stays narrow and does not link worktrees to
claims or enrich the state brief.
Endpoints:
POST /api/v5/worktrees - create a git worktree
GET /api/v5/worktrees?space_id=... - list worktrees for a space
GET /api/v5/worktrees/{worktree_id} - show one worktree row
DELETE /api/v5/worktrees/{worktree_id} - close/remove a worktree
"""
from __future__ import annotations
import re
import shutil
import subprocess
import uuid
from datetime import UTC, datetime
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db.database import get_db
from app.db.models import CommitModel, RepoModel, WorkTree
router = APIRouter(prefix="/worktrees", tags=["worktrees-v5"])
DEMO_USER_ID = uuid.UUID("00000000-0000-0000-0000-00000000000a")
VALID_STATUSES = {"active", "closed"}
def _utcnow() -> datetime:
return datetime.now(UTC)
def _slugify(value: str, fallback: str = "item") -> str:
"""Lowercase a user/project label into a filesystem/branch-safe slug."""
slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
return slug or fallback
def _default_branch_name(agent: str, suffix: str) -> str:
return f"smriti/{_slugify(agent, fallback='agent')}/{suffix}"
def _default_worktree_path(space_name: str, agent: str, suffix: str) -> str:
space_slug = _slugify(space_name, fallback="space")
agent_slug = _slugify(agent, fallback="agent")
return str(
Path.home()
/ ".smriti"
/ "worktrees"
/ space_slug
/ f"{agent_slug}-{suffix}"
)
def _run_git(
args: list[str],
*,
cwd: str | Path | None = None,
timeout: float = 30.0,
) -> subprocess.CompletedProcess[str]:
"""Run git with captured output and consistent infrastructure errors."""
try:
return subprocess.run(
["git", *args],
cwd=str(cwd) if cwd is not None else None,
capture_output=True,
text=True,
check=False,
timeout=timeout,
)
except FileNotFoundError:
raise HTTPException(status_code=500, detail="git executable not found")
except subprocess.TimeoutExpired:
raise HTTPException(status_code=500, detail="git command timed out")
def _get_repo(space_id: uuid.UUID, db: Session) -> RepoModel:
repo = db.get(RepoModel, space_id)
if not repo or repo.user_id != DEMO_USER_ID:
raise HTTPException(status_code=404, detail="Space not found")
return repo
def _latest_project_root(space_id: uuid.UUID, db: Session) -> Path:
stmt = (
select(CommitModel.project_root)
.where(
CommitModel.repo_id == space_id,
CommitModel.project_root.is_not(None),
CommitModel.project_root != "",
)
.order_by(CommitModel.created_at.desc())
.limit(1)
)
project_root = db.scalars(stmt).first()
if not project_root:
raise HTTPException(
status_code=400,
detail=(
"No checkpoint with project_root found for this space; "
"create a checkpoint from the project checkout first."
),
)
resolved = Path(project_root).expanduser().resolve()
if not resolved.is_dir():
raise HTTPException(
status_code=400,
detail=f"project_root does not exist on disk: {resolved}",
)
return resolved
def _current_head(project_root: Path) -> str:
result = _run_git(["rev-parse", "HEAD"], cwd=project_root)
if result.returncode != 0:
detail = result.stderr.strip() or result.stdout.strip() or "could not resolve HEAD"
raise HTTPException(status_code=500, detail=detail)
return result.stdout.strip()
def _branch_exists(project_root: Path, branch_name: str) -> bool:
result = _run_git(
["show-ref", "--verify", "--quiet", f"refs/heads/{branch_name}"],
cwd=project_root,
)
if result.returncode == 0:
return True
if result.returncode == 1:
return False
detail = result.stderr.strip() or "could not check branch existence"
raise HTTPException(status_code=500, detail=detail)
def _resolve_target_path(path_value: str) -> Path:
target = Path(path_value).expanduser()
if not target.is_absolute():
raise HTTPException(status_code=400, detail="base_path must be absolute")
return target.resolve()
def _git_error(result: subprocess.CompletedProcess[str], fallback: str) -> str:
return result.stderr.strip() or result.stdout.strip() or fallback
# -- Request / Response schemas ----------------------------------------------
class CreateWorkTreeRequest(BaseModel):
space_id: str
agent: str
branch_name: str | None = None
base_commit_sha: str | None = None
base_path: str | None = Field(
default=None,
description="Absolute target path for the new worktree.",
)
class WorkTreeResponse(BaseModel):
id: uuid.UUID
repo_id: uuid.UUID
agent: str
path: str
branch_name: str
base_commit_sha: str | None = None
status: str
created_at: datetime
closed_at: datetime | None = None
model_config = {"from_attributes": True}
# -- Endpoints ----------------------------------------------------------------
@router.post("", response_model=WorkTreeResponse, status_code=201)
def create_worktree(
payload: CreateWorkTreeRequest,
db: Session = Depends(get_db),
):
"""Create a git worktree and record it after git succeeds."""
try:
space_id = uuid.UUID(payload.space_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid space_id")
repo = _get_repo(space_id, db)
agent = payload.agent.strip()
if not agent:
raise HTTPException(status_code=400, detail="agent must be non-empty")
project_root = _latest_project_root(space_id, db)
suffix = uuid.uuid4().hex[:8]
branch_name = (payload.branch_name or "").strip() or _default_branch_name(agent, suffix)
target_path = _resolve_target_path(
payload.base_path or _default_worktree_path(repo.name, agent, suffix)
)
if _branch_exists(project_root, branch_name):
raise HTTPException(
status_code=409,
detail=f"Branch already exists: {branch_name}",
)
base_commit_sha = (payload.base_commit_sha or "").strip() or _current_head(project_root)
target_existed = target_path.exists()
target_path.parent.mkdir(parents=True, exist_ok=True)
result = _run_git(
[
"worktree",
"add",
"-b",
branch_name,
str(target_path),
base_commit_sha,
],
cwd=project_root,
)
if result.returncode != 0:
if target_path.exists() and not target_existed:
shutil.rmtree(target_path, ignore_errors=True)
raise HTTPException(
status_code=500,
detail=_git_error(result, "git worktree add failed"),
)
worktree = WorkTree(
repo_id=space_id,
agent=agent,
path=str(target_path),
branch_name=branch_name,
base_commit_sha=base_commit_sha,
status="active",
)
db.add(worktree)
db.commit()
db.refresh(worktree)
return worktree
@router.get("", response_model=list[WorkTreeResponse])
def list_worktrees(
space_id: uuid.UUID = Query(..., description="Space UUID"),
include_closed: bool = Query(False, description="Include closed worktrees"),
db: Session = Depends(get_db),
):
"""List worktrees for a space, newest first."""
_get_repo(space_id, db)
stmt = (
select(WorkTree)
.where(WorkTree.repo_id == space_id)
.order_by(WorkTree.created_at.desc())
)
if not include_closed:
stmt = stmt.where(WorkTree.status == "active")
return list(db.scalars(stmt).all())
@router.get("/{worktree_id}", response_model=WorkTreeResponse)
def get_worktree(worktree_id: uuid.UUID, db: Session = Depends(get_db)):
"""Return one worktree row."""
worktree = db.get(WorkTree, worktree_id)
if not worktree:
raise HTTPException(status_code=404, detail="Worktree not found")
return worktree
@router.delete("/{worktree_id}", response_model=WorkTreeResponse)
def close_worktree(
worktree_id: uuid.UUID,
force: bool = Query(False, description="Force removal even if dirty"),
db: Session = Depends(get_db),
):
"""Remove a git worktree and mark its row closed."""
worktree = db.get(WorkTree, worktree_id)
if not worktree:
raise HTTPException(status_code=404, detail="Worktree not found")
if worktree.status == "closed":
raise HTTPException(status_code=409, detail="Worktree is already closed")
if worktree.status not in VALID_STATUSES:
raise HTTPException(
status_code=409,
detail=f"Worktree has invalid status: {worktree.status}",
)
project_root = _latest_project_root(worktree.repo_id, db)
worktree_path = Path(worktree.path).expanduser().resolve()
if not force:
status = _run_git(["status", "--porcelain"], cwd=worktree_path)
if status.returncode != 0:
raise HTTPException(
status_code=500,
detail=_git_error(status, "git status failed"),
)
if status.stdout.strip():
raise HTTPException(
status_code=409,
detail="Worktree has uncommitted changes; pass force=true to remove it.",
)
args = ["worktree", "remove"]
if force:
args.append("--force")
args.append(str(worktree_path))
result = _run_git(args, cwd=project_root)
if result.returncode != 0:
raise HTTPException(
status_code=500,
detail=_git_error(result, "git worktree remove failed"),
)
worktree.status = "closed"
worktree.closed_at = _utcnow()
db.commit()
db.refresh(worktree)
return worktree

View file

@ -146,6 +146,9 @@ class RepoModel(Base):
commits: Mapped[list["CommitModel"]] = relationship( commits: Mapped[list["CommitModel"]] = relationship(
back_populates="repo", cascade="all, delete-orphan", order_by="CommitModel.created_at.desc()" back_populates="repo", cascade="all, delete-orphan", order_by="CommitModel.created_at.desc()"
) )
worktrees: Mapped[list["WorkTree"]] = relationship(
back_populates="repo", cascade="all, delete-orphan"
)
class CommitModel(Base): class CommitModel(Base):
__tablename__ = "commits" __tablename__ = "commits"
@ -314,3 +317,28 @@ class WorkClaim(Base):
DateTime(timezone=True), nullable=False, DateTime(timezone=True), nullable=False,
) )
class WorkTree(Base):
"""A git worktree allocated for an agent working in a space.
Worktrees provide filesystem-level isolation: separate working
directories and git indexes backed by the same repository object store.
"""
__tablename__ = "work_trees"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
repo_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("repos.id", ondelete="CASCADE"),
nullable=False, index=True,
)
agent: Mapped[str] = mapped_column(String(100), nullable=False)
path: Mapped[str] = mapped_column(Text, nullable=False)
branch_name: Mapped[str] = mapped_column(String(255), nullable=False)
base_commit_sha: Mapped[str | None] = mapped_column(String(64), nullable=True)
status: Mapped[str] = mapped_column(String(20), default="active")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
repo: Mapped["RepoModel"] = relationship(back_populates="worktrees")

View file

@ -2,6 +2,7 @@ import logging
import pathlib import pathlib
import re import re
import subprocess import subprocess
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
@ -52,7 +53,17 @@ def create_app() -> FastAPI:
for handler in logging.root.handlers: for handler in logging.root.handlers:
handler.addFilter(SecretGuardFilter()) handler.addFilter(SecretGuardFilter())
from app.api.routes import repos, commits, context_git, chat, checkpoint, lineage, claims, metrics from app.api.routes import (
chat,
checkpoint,
claims,
commits,
context_git,
lineage,
metrics,
repos,
worktrees,
)
# V2 Routes (Git for memory) # V2 Routes (Git for memory)
app.include_router(repos.router, prefix="/api/v2", tags=["repos"]) app.include_router(repos.router, prefix="/api/v2", tags=["repos"])
@ -67,6 +78,7 @@ def create_app() -> FastAPI:
app.include_router(lineage.router, prefix="/api/v5", tags=["lineage-v5"]) app.include_router(lineage.router, prefix="/api/v5", tags=["lineage-v5"])
app.include_router(claims.router, prefix="/api/v5", tags=["claims-v5"]) app.include_router(claims.router, prefix="/api/v5", tags=["claims-v5"])
app.include_router(metrics.router, prefix="/api/v5", tags=["metrics-v5"]) app.include_router(metrics.router, prefix="/api/v5", tags=["metrics-v5"])
app.include_router(worktrees.router, prefix="/api/v5", tags=["worktrees-v5"])
# ── Capabilities manifest ──────────────────────────────────────── # ── Capabilities manifest ────────────────────────────────────────
# Computed once at startup so /health is zero-cost at request time. # Computed once at startup so /health is zero-cost at request time.
@ -83,6 +95,7 @@ def create_app() -> FastAPI:
"branch_disposition", # PATCH /api/v5/lineage/branches/disposition "branch_disposition", # PATCH /api/v5/lineage/branches/disposition
"freshness", # since_commit_id on state endpoint "freshness", # since_commit_id on state endpoint
"compact_state", # --compact mode on state brief "compact_state", # --compact mode on state brief
"worktrees", # /api/v5/worktrees
] ]
@app.get("/health") @app.get("/health")

View file

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

View file

@ -0,0 +1,266 @@
"""Integration tests for the V5 worktree API.
These tests use a real temporary git repository on disk. That is the
important safety boundary for this feature: V1 is about filesystem/index
isolation, so the API must exercise real `git worktree` behavior.
"""
from __future__ import annotations
import subprocess
import uuid
from pathlib import Path
from types import SimpleNamespace
from fastapi import HTTPException
from app.api.routes import worktrees
def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
result = subprocess.run(
["git", *args],
cwd=repo,
capture_output=True,
text=True,
check=False,
)
if check:
assert result.returncode == 0, result.stderr
return result
def _create_repo(client, name="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_session(client, repo_id, title="worktree 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_with_root(client, repo_id, session_id, project_root: Path):
payload = {
"repo_id": repo_id,
"session_id": session_id,
"message": "base",
"summary": "base",
"project_root": str(project_root),
}
r = client.post("/api/v4/chat/commit", json=payload)
assert r.status_code == 201, r.text
return r.json()
def _create_project_with_root(client, git_repo: Path):
repo_id = _create_repo(client)
session_id = _create_session(client, repo_id)
_commit_with_root(client, repo_id, session_id, git_repo)
return repo_id
def _create_worktree(client, space_id: str, **kwargs):
payload = {
"space_id": space_id,
"agent": kwargs.pop("agent", "codex-local"),
**kwargs,
}
return client.post("/api/v5/worktrees", json=payload)
def test_create_worktree_inserts_row_creates_directory_and_branch(
client,
tmp_path,
monkeypatch,
):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
monkeypatch.setattr(worktrees.Path, "home", lambda: tmp_path / "home")
r = _create_worktree(client, space_id, agent="Codex Local")
assert r.status_code == 201, r.text
data = r.json()
worktree_path = Path(data["path"])
assert worktree_path.exists()
assert data["agent"] == "Codex Local"
assert data["branch_name"].startswith("smriti/codex-local/")
assert data["base_commit_sha"] == _git(git_repo, "rev-parse", "HEAD").stdout.strip()
assert data["status"] == "active"
assert _git(
git_repo,
"show-ref",
"--verify",
f"refs/heads/{data['branch_name']}",
).returncode == 0
def test_create_worktree_honors_explicit_branch_base_and_path(client, tmp_path):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
base_sha = _git(git_repo, "rev-parse", "HEAD").stdout.strip()
target = tmp_path / "custom-worktree"
r = _create_worktree(
client,
space_id,
branch_name="feature/custom-worktree",
base_commit_sha=base_sha,
base_path=str(target),
)
assert r.status_code == 201, r.text
data = r.json()
assert data["branch_name"] == "feature/custom-worktree"
assert data["base_commit_sha"] == base_sha
assert data["path"] == str(target.resolve())
assert target.exists()
def test_create_worktree_existing_branch_rejected_without_row_or_directory(
client,
tmp_path,
):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
_git(git_repo, "branch", "existing-worktree-branch")
target = tmp_path / "should-not-exist"
r = _create_worktree(
client,
space_id,
branch_name="existing-worktree-branch",
base_path=str(target),
)
assert r.status_code == 409
assert "Branch already exists" in r.json()["detail"]
assert not target.exists()
assert client.get(f"/api/v5/worktrees?space_id={space_id}").json() == []
def test_list_show_and_close_clean_worktree(client, tmp_path):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
target = tmp_path / "list-show-close"
created = _create_worktree(client, space_id, base_path=str(target)).json()
list_r = client.get(f"/api/v5/worktrees?space_id={space_id}")
assert list_r.status_code == 200
assert [w["id"] for w in list_r.json()] == [created["id"]]
show_r = client.get(f"/api/v5/worktrees/{created['id']}")
assert show_r.status_code == 200
assert show_r.json()["path"] == str(target.resolve())
close_r = client.delete(f"/api/v5/worktrees/{created['id']}")
assert close_r.status_code == 200, close_r.text
closed = close_r.json()
assert closed["status"] == "closed"
assert closed["closed_at"] is not None
assert not target.exists()
active_r = client.get(f"/api/v5/worktrees?space_id={space_id}")
assert active_r.status_code == 200
assert active_r.json() == []
all_r = client.get(f"/api/v5/worktrees?space_id={space_id}&include_closed=true")
assert len(all_r.json()) == 1
def test_show_nonexistent_worktree_returns_404(client):
r = client.get(f"/api/v5/worktrees/{uuid.uuid4()}")
assert r.status_code == 404
def test_close_dirty_worktree_requires_force(client, tmp_path):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
target = tmp_path / "dirty-worktree"
created = _create_worktree(client, space_id, base_path=str(target)).json()
(target / "dirty.txt").write_text("uncommitted\n")
r = client.delete(f"/api/v5/worktrees/{created['id']}")
assert r.status_code == 409
assert "uncommitted changes" in r.json()["detail"]
assert target.exists()
force_r = client.delete(f"/api/v5/worktrees/{created['id']}?force=true")
assert force_r.status_code == 200, force_r.text
assert force_r.json()["status"] == "closed"
assert not target.exists()
def test_close_already_closed_returns_409(client, tmp_path):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
created = _create_worktree(
client,
space_id,
base_path=str(tmp_path / "already-closed"),
).json()
first = client.delete(f"/api/v5/worktrees/{created['id']}")
assert first.status_code == 200
second = client.delete(f"/api/v5/worktrees/{created['id']}")
assert second.status_code == 409
assert "already closed" in second.json()["detail"]
def test_create_worktree_requires_project_root(client, tmp_path):
space_id = _create_repo(client)
target = tmp_path / "no-root-worktree"
r = _create_worktree(client, space_id, base_path=str(target))
assert r.status_code == 400
assert "project_root" in r.json()["detail"]
assert not target.exists()
def test_git_infrastructure_error_does_not_write_row_or_leave_target(
client,
tmp_path,
monkeypatch,
):
git_repo = _init_git_repo(tmp_path)
space_id = _create_project_with_root(client, git_repo)
target = tmp_path / "git-missing"
head_sha = _git(git_repo, "rev-parse", "HEAD").stdout.strip()
def fake_run_git(args, *, cwd=None, timeout=30.0):
if args[0] == "show-ref":
return SimpleNamespace(returncode=1, stdout="", stderr="")
if args[0] == "rev-parse":
return SimpleNamespace(returncode=0, stdout=f"{head_sha}\n", stderr="")
if args[:2] == ["worktree", "add"]:
raise HTTPException(status_code=500, detail="git executable not found")
raise AssertionError(f"unexpected git call: {args}")
monkeypatch.setattr(worktrees, "_run_git", fake_run_git)
r = _create_worktree(client, space_id, base_path=str(target))
assert r.status_code == 500
assert "git executable not found" in r.json()["detail"]
assert not target.exists()
assert client.get(f"/api/v5/worktrees?space_id={space_id}").json() == []
def _init_git_repo(tmp_path: Path) -> Path:
repo = tmp_path / "project"
repo.mkdir()
_git(repo, "init")
_git(repo, "config", "user.email", "test@example.com")
_git(repo, "config", "user.name", "Test User")
(repo / "README.md").write_text("hello\n")
_git(repo, "add", "README.md")
_git(repo, "commit", "-m", "initial")
return repo

View file

@ -0,0 +1,33 @@
"""Unit tests for worktree naming/path helpers."""
from __future__ import annotations
from pathlib import Path
from app.api.routes import worktrees
def test_slugify_lowercases_and_replaces_non_alphanumeric():
assert worktrees._slugify("Smriti Dev!") == "smriti-dev"
assert worktrees._slugify("Claude_Code v2") == "claude-code-v2"
assert worktrees._slugify("!!!", fallback="fallback") == "fallback"
def test_default_branch_name_uses_agent_slug():
assert (
worktrees._default_branch_name("Claude Code", "abc12345")
== "smriti/claude-code/abc12345"
)
def test_default_worktree_path_uses_home_space_agent_and_suffix(monkeypatch, tmp_path):
monkeypatch.setattr(worktrees.Path, "home", lambda: Path(tmp_path))
path = worktrees._default_worktree_path(
"Smriti Dev",
"Codex Local",
"abc12345",
)
assert path == str(
tmp_path / ".smriti" / "worktrees" / "smriti-dev" / "codex-local-abc12345"
)

View file

@ -212,6 +212,45 @@ class SmritiClient:
params["include_expired"] = "true" params["include_expired"] = "true"
return self._request("GET", "/api/v5/claims", params=params) return self._request("GET", "/api/v5/claims", params=params)
# ── Worktrees ───────────────────────────────────────────────────
def create_worktree(
self,
space_id: str,
agent: str,
branch_name: str | None = None,
base_commit_sha: str | None = None,
base_path: str | None = None,
) -> dict:
"""POST /api/v5/worktrees — create a git worktree."""
payload = {
"space_id": space_id,
"agent": agent,
}
if branch_name:
payload["branch_name"] = branch_name
if base_commit_sha:
payload["base_commit_sha"] = base_commit_sha
if base_path:
payload["base_path"] = base_path
return self._request("POST", "/api/v5/worktrees", json=payload)
def list_worktrees(self, space_id: str, include_closed: bool = False) -> list[dict]:
"""GET /api/v5/worktrees?space_id=... — list worktrees."""
params = {"space_id": space_id}
if include_closed:
params["include_closed"] = "true"
return self._request("GET", "/api/v5/worktrees", params=params)
def get_worktree(self, worktree_id: str) -> dict:
"""GET /api/v5/worktrees/{id} — show one worktree."""
return self._request("GET", f"/api/v5/worktrees/{worktree_id}")
def close_worktree(self, worktree_id: str, force: bool = False) -> dict:
"""DELETE /api/v5/worktrees/{id} — close/remove a worktree."""
params = {"force": "true"} if force else None
return self._request("DELETE", f"/api/v5/worktrees/{worktree_id}", params=params)
# ── Metrics ───────────────────────────────────────────────────── # ── Metrics ─────────────────────────────────────────────────────
def get_space_metrics(self, space_id: str) -> dict: def get_space_metrics(self, space_id: str) -> dict:

View file

@ -18,6 +18,10 @@ Commands for agent and programmatic use:
smriti checkpoint list <space> smriti checkpoint list <space>
smriti checkpoint review <checkpoint-id> smriti checkpoint review <checkpoint-id>
smriti checkpoint delete <checkpoint-id> [--cascade] [-y] smriti checkpoint delete <checkpoint-id> [--cascade] [-y]
smriti worktree open <space> --agent <id>
smriti worktree list <space>
smriti worktree show <worktree-id>
smriti worktree close <worktree-id>
Multi-branch workflow: use `smriti fork <checkpoint>` to start a new Multi-branch workflow: use `smriti fork <checkpoint>` to start a new
session on a new branch, then `smriti checkpoint create <space> --session session on a new branch, then `smriti checkpoint create <space> --session
@ -804,6 +808,99 @@ def cmd_claim_list(client: SmritiClient, args: argparse.Namespace) -> None:
print(f" - `{agent}` [{intent}] on `{branch}` — {scope} (id: {c['id']})") print(f" - `{agent}` [{intent}] on `{branch}` — {scope} (id: {c['id']})")
# ── worktree subcommand handlers ────────────────────────────────────────────
def _short_id(value: str) -> str:
return f"{value[:8]}" if len(value) > 8 else value
def _display_path(path: str) -> str:
home = os.path.expanduser("~")
if path == home:
return "~"
if path.startswith(home + os.sep):
return "~" + path[len(home):]
return path
def _print_worktree_table(worktrees: list[dict]) -> None:
headers = ["ID", "AGENT", "BRANCH", "DIRTY", "AHEAD", "PATH"]
rows = [
[
_short_id(str(w.get("id", ""))),
str(w.get("agent", "")),
str(w.get("branch_name", "")),
"",
"",
_display_path(str(w.get("path", ""))),
]
for w in worktrees
]
widths = [
max(len(headers[i]), *(len(row[i]) for row in rows))
for i in range(len(headers))
]
print(" ".join(headers[i].ljust(widths[i]) for i in range(len(headers))))
for row in rows:
print(" ".join(row[i].ljust(widths[i]) for i in range(len(row))))
def cmd_worktree_open(client: SmritiClient, args: argparse.Namespace) -> None:
"""Create a git worktree for an agent and print its path."""
space = client.resolve_space(args.space)
worktree = client.create_worktree(
space_id=space["id"],
agent=args.agent,
branch_name=args.branch,
base_commit_sha=args.base_commit,
base_path=args.base_path,
)
if args.json:
_print_json(worktree)
else:
print(worktree["path"])
def cmd_worktree_list(client: SmritiClient, args: argparse.Namespace) -> None:
"""List worktrees for a space."""
space = client.resolve_space(args.space)
worktrees = client.list_worktrees(space["id"], include_closed=args.include_closed)
if args.json:
_print_json(worktrees)
return
if not worktrees:
print("No worktrees." if args.include_closed else "No active worktrees.")
return
_print_worktree_table(worktrees)
def cmd_worktree_show(client: SmritiClient, args: argparse.Namespace) -> None:
"""Show one worktree."""
worktree = client.get_worktree(args.worktree_id)
if args.json:
_print_json(worktree)
return
print(f"id: {worktree['id']}")
print(f"status: {worktree['status']}")
print(f"agent: {worktree['agent']}")
print(f"branch: {worktree['branch_name']}")
print(f"base_commit: {worktree.get('base_commit_sha') or ''}")
print(f"path: {worktree['path']}")
print(f"created_at: {worktree['created_at']}")
if worktree.get("closed_at"):
print(f"closed_at: {worktree['closed_at']}")
def cmd_worktree_close(client: SmritiClient, args: argparse.Namespace) -> None:
"""Close/remove a git worktree."""
worktree = client.close_worktree(args.worktree_id, force=args.force)
if args.json:
_print_json(worktree)
else:
print(f"Closed worktree `{worktree['id']}` at {worktree['path']}.")
def cmd_restore(client: SmritiClient, args: argparse.Namespace) -> None: def cmd_restore(client: SmritiClient, args: argparse.Namespace) -> None:
commit = client.get_commit(args.checkpoint_id) commit = client.get_commit(args.checkpoint_id)
space = client.get_space(str(commit.get("repo_id", ""))) space = client.get_space(str(commit.get("repo_id", "")))
@ -1147,6 +1244,39 @@ def _build_parser() -> argparse.ArgumentParser:
cl_list.add_argument("--json", action="store_true") cl_list.add_argument("--json", action="store_true")
cl_list.set_defaults(func=cmd_claim_list) cl_list.set_defaults(func=cmd_claim_list)
# worktree — filesystem/index isolation for agent work
worktree_parser = subparsers.add_parser(
"worktree",
help="Manage git worktrees for isolated agent working directories",
)
worktree_sub = worktree_parser.add_subparsers(dest="subcommand", required=True)
wt_open = worktree_sub.add_parser("open", help="Create a worktree for an agent")
wt_open.add_argument("space", help="Space name or UUID")
wt_open.add_argument("--agent", required=True, help="Agent identifier (e.g. claude-code)")
wt_open.add_argument("--branch", help="Branch name for the new worktree", default=None)
wt_open.add_argument("--base-commit", dest="base_commit", help="Git SHA to base the worktree on", default=None)
wt_open.add_argument("--base-path", dest="base_path", help="Absolute target path for the new worktree", default=None)
wt_open.add_argument("--json", action="store_true")
wt_open.set_defaults(func=cmd_worktree_open)
wt_list = worktree_sub.add_parser("list", help="List worktrees for a space")
wt_list.add_argument("space", help="Space name or UUID")
wt_list.add_argument("--include-closed", action="store_true", help="Include closed worktrees")
wt_list.add_argument("--json", action="store_true")
wt_list.set_defaults(func=cmd_worktree_list)
wt_show = worktree_sub.add_parser("show", help="Show one worktree")
wt_show.add_argument("worktree_id", help="Worktree UUID")
wt_show.add_argument("--json", action="store_true")
wt_show.set_defaults(func=cmd_worktree_show)
wt_close = worktree_sub.add_parser("close", help="Close/remove a worktree")
wt_close.add_argument("worktree_id", help="Worktree UUID")
wt_close.add_argument("--force", action="store_true", help="Force removal even if dirty")
wt_close.add_argument("--json", action="store_true")
wt_close.set_defaults(func=cmd_worktree_close)
# skills — install the Smriti agent skill pack into an agent host's # skills — install the Smriti agent skill pack into an agent host's
# project directory. The skill pack teaches agents when and why to # project directory. The skill pack teaches agents when and why to
# use Smriti's tools (the load-bearing anti-pattern section is in # use Smriti's tools (the load-bearing anti-pattern section is in

View file

@ -1,6 +1,6 @@
"""Smriti MCP server — stdio transport. """Smriti MCP server — stdio transport.
Exposes the Smriti CLI surface as 12 MCP tools so agents inside MCP-aware Exposes the Smriti CLI surface as 21 MCP tools so agents inside MCP-aware
hosts (Claude Code, Cursor, Windsurf) can read and write reasoning state hosts (Claude Code, Cursor, Windsurf) can read and write reasoning state
without shelling out to the `smriti` binary. without shelling out to the `smriti` binary.
@ -616,6 +616,138 @@ def smriti_claim_done(claim_id: str, abandon: bool = False) -> str:
return f"Claim `{claim['id']}` marked {claim['status']}." return f"Claim `{claim['id']}` marked {claim['status']}."
def _format_worktree(worktree: dict) -> str:
lines = [
f"id: {worktree['id']}",
f"status: {worktree['status']}",
f"agent: {worktree['agent']}",
f"branch: {worktree['branch_name']}",
f"base_commit: {worktree.get('base_commit_sha') or ''}",
f"path: {worktree['path']}",
f"created_at: {worktree['created_at']}",
]
if worktree.get("closed_at"):
lines.append(f"closed_at: {worktree['closed_at']}")
return "\n".join(lines)
def _format_worktree_list(worktrees: list[dict]) -> str:
if not worktrees:
return "No worktrees."
lines = ["| ID | Agent | Branch | Dirty | Ahead | Path |", "|---|---|---|---|---|---|"]
for worktree in worktrees:
short_id = str(worktree.get("id", ""))[:8]
lines.append(
"| "
f"{short_id} | "
f"{worktree.get('agent', '')} | "
f"{worktree.get('branch_name', '')} | "
"— | "
"— | "
f"{worktree.get('path', '')} |"
)
return "\n".join(lines)
@mcp.tool()
def smriti_worktree_open(
space: str,
agent: str = "claude-code",
branch: str = "",
base_commit: str = "",
base_path: str = "",
) -> str:
"""Create a git worktree for isolated agent work.
Use this when an agent is about to work in a repository and needs a
separate filesystem path and git index instead of sharing the main
checkout. V1 only creates the worktree and records it; it does not
bind the worktree to a claim or change the state brief.
Args:
space: Space name or UUID.
agent: Agent identifier (default: claude-code).
branch: Optional branch name for the new worktree.
base_commit: Optional git SHA to base the worktree on. Default:
current HEAD of the project repo.
base_path: Optional absolute target path. Default:
~/.smriti/worktrees/<space>/<agent>-<suffix>.
"""
client = _client()
try:
s = client.resolve_space(space)
worktree = client.create_worktree(
space_id=s["id"],
agent=agent,
branch_name=branch or None,
base_commit_sha=base_commit or None,
base_path=base_path or None,
)
except SmritiError as e:
_raise_from(e)
return (
f"Created worktree `{worktree['id']}` for `{worktree['agent']}`.\n"
f"Path: `{worktree['path']}`\n"
f"Branch: `{worktree['branch_name']}`\n"
f"Base: `{worktree.get('base_commit_sha') or ''}`"
)
@mcp.tool()
def smriti_worktree_list(space: str, include_closed: bool = False) -> str:
"""List worktrees for a Smriti space.
Use this to see active agent worktree directories. Dirty/ahead columns
are placeholders in V1 and always render as unknown until a future git
status enrichment pass lands.
Args:
space: Space name or UUID.
include_closed: Include closed worktrees. Default False.
"""
client = _client()
try:
s = client.resolve_space(space)
worktrees = client.list_worktrees(s["id"], include_closed=include_closed)
except SmritiError as e:
_raise_from(e)
return _format_worktree_list(worktrees)
@mcp.tool()
def smriti_worktree_show(worktree_id: str) -> str:
"""Show one worktree row by UUID.
Args:
worktree_id: Worktree UUID returned by smriti_worktree_open or
smriti_worktree_list.
"""
try:
worktree = _client().get_worktree(worktree_id)
except SmritiError as e:
_raise_from(e)
return _format_worktree(worktree)
@mcp.tool()
def smriti_worktree_close(worktree_id: str, force: bool = False) -> str:
"""Close and remove a git worktree.
By default this refuses to remove a dirty worktree. Pass force=True
only when the uncommitted contents are intentionally disposable.
Args:
worktree_id: Worktree UUID returned by smriti_worktree_open or
smriti_worktree_list.
force: Force removal even if the worktree is dirty.
"""
try:
worktree = _client().close_worktree(worktree_id, force=force)
except SmritiError as e:
_raise_from(e)
return f"Closed worktree `{worktree['id']}` at `{worktree['path']}`."
@mcp.tool() @mcp.tool()
def smriti_install_skill(target: str) -> str: def smriti_install_skill(target: str) -> str:
"""Return the Smriti agent skill pack for an agent target. """Return the Smriti agent skill pack for an agent target.

View file

@ -0,0 +1,150 @@
from __future__ import annotations
import argparse
from unittest.mock import MagicMock
import pytest
from smriti_cli import main as cli_main
from smriti_cli.client import SmritiClient
def _worktree_dict(**overrides):
base = {
"id": "11111111-2222-3333-4444-555555555555",
"repo_id": "space-uuid",
"agent": "claude-1",
"path": "/tmp/worktree",
"branch_name": "smriti/claude-1/abc12345",
"base_commit_sha": "abc123",
"status": "active",
"created_at": "2026-05-04T00:00:00Z",
"closed_at": None,
}
base.update(overrides)
return base
def test_worktree_open_parser_wiring():
parser = cli_main._build_parser()
args = parser.parse_args(
[
"worktree",
"open",
"my-project",
"--agent",
"claude-1",
"--branch",
"feature/wt",
"--base-commit",
"abc123",
"--base-path",
"/tmp/wt",
]
)
assert args.command == "worktree"
assert args.subcommand == "open"
assert args.space == "my-project"
assert args.agent == "claude-1"
assert args.branch == "feature/wt"
assert args.base_commit == "abc123"
assert args.base_path == "/tmp/wt"
assert args.func is cli_main.cmd_worktree_open
def test_cmd_worktree_open_calls_client_and_prints_path(capsys: pytest.CaptureFixture[str]):
client = MagicMock(spec=SmritiClient)
client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
client.create_worktree.return_value = _worktree_dict(path="/tmp/wt")
args = argparse.Namespace(
space="my-project",
agent="claude-1",
branch=None,
base_commit=None,
base_path=None,
json=False,
)
cli_main.cmd_worktree_open(client, args)
out = capsys.readouterr().out
assert out == "/tmp/wt\n"
client.resolve_space.assert_called_once_with("my-project")
client.create_worktree.assert_called_once_with(
space_id="space-uuid",
agent="claude-1",
branch_name=None,
base_commit_sha=None,
base_path=None,
)
def test_cmd_worktree_open_json_path():
client = MagicMock(spec=SmritiClient)
client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
worktree = _worktree_dict(path="/tmp/json-wt")
client.create_worktree.return_value = worktree
args = argparse.Namespace(
space="my-project",
agent="claude-1",
branch="feature/wt",
base_commit="abc123",
base_path="/tmp/json-wt",
json=True,
)
captured: list[dict] = []
original = cli_main._print_json
cli_main._print_json = captured.append
try:
cli_main.cmd_worktree_open(client, args)
finally:
cli_main._print_json = original
assert captured == [worktree]
client.create_worktree.assert_called_once_with(
space_id="space-uuid",
agent="claude-1",
branch_name="feature/wt",
base_commit_sha="abc123",
base_path="/tmp/json-wt",
)
def test_cmd_worktree_list_calls_client(capsys: pytest.CaptureFixture[str]):
client = MagicMock(spec=SmritiClient)
client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
client.list_worktrees.return_value = [_worktree_dict()]
args = argparse.Namespace(space="my-project", include_closed=False, json=False)
cli_main.cmd_worktree_list(client, args)
out = capsys.readouterr().out
assert "AGENT" in out
assert "claude-1" in out
client.list_worktrees.assert_called_once_with("space-uuid", include_closed=False)
def test_cmd_worktree_show_calls_client(capsys: pytest.CaptureFixture[str]):
client = MagicMock(spec=SmritiClient)
client.get_worktree.return_value = _worktree_dict()
args = argparse.Namespace(worktree_id="wt-uuid", json=False)
cli_main.cmd_worktree_show(client, args)
out = capsys.readouterr().out
assert "branch: smriti/claude-1/abc12345" in out
client.get_worktree.assert_called_once_with("wt-uuid")
def test_cmd_worktree_close_calls_client(capsys: pytest.CaptureFixture[str]):
client = MagicMock(spec=SmritiClient)
client.close_worktree.return_value = _worktree_dict(status="closed")
args = argparse.Namespace(worktree_id="wt-uuid", force=True, json=False)
cli_main.cmd_worktree_close(client, args)
out = capsys.readouterr().out
assert "Closed worktree" in out
client.close_worktree.assert_called_once_with("wt-uuid", force=True)

View file

@ -0,0 +1,77 @@
from __future__ import annotations
from smriti_cli import mcp_server
def _worktree_dict(**overrides):
base = {
"id": "11111111-2222-3333-4444-555555555555",
"repo_id": "space-uuid",
"agent": "claude-1",
"path": "/tmp/worktree",
"branch_name": "smriti/claude-1/abc12345",
"base_commit_sha": "abc123",
"status": "active",
"created_at": "2026-05-04T00:00:00Z",
"closed_at": None,
}
base.update(overrides)
return base
def test_mcp_worktree_open_calls_client(mock_client):
mock_client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
mock_client.create_worktree.return_value = _worktree_dict(path="/tmp/wt")
result = mcp_server.smriti_worktree_open(
space="my-project",
agent="claude-1",
branch="feature/wt",
base_commit="abc123",
base_path="/tmp/wt",
)
assert "/tmp/wt" in result
mock_client.resolve_space.assert_called_once_with("my-project")
mock_client.create_worktree.assert_called_once_with(
space_id="space-uuid",
agent="claude-1",
branch_name="feature/wt",
base_commit_sha="abc123",
base_path="/tmp/wt",
)
def test_mcp_worktree_list_calls_client(mock_client):
mock_client.resolve_space.return_value = {"id": "space-uuid", "name": "my-project"}
mock_client.list_worktrees.return_value = [_worktree_dict()]
result = mcp_server.smriti_worktree_list(
space="my-project",
include_closed=True,
)
assert "claude-1" in result
mock_client.resolve_space.assert_called_once_with("my-project")
mock_client.list_worktrees.assert_called_once_with(
"space-uuid",
include_closed=True,
)
def test_mcp_worktree_show_calls_client(mock_client):
mock_client.get_worktree.return_value = _worktree_dict()
result = mcp_server.smriti_worktree_show("wt-uuid")
assert "smriti/claude-1/abc12345" in result
mock_client.get_worktree.assert_called_once_with("wt-uuid")
def test_mcp_worktree_close_calls_client(mock_client):
mock_client.close_worktree.return_value = _worktree_dict(status="closed")
result = mcp_server.smriti_worktree_close("wt-uuid", force=True)
assert "Closed worktree" in result
mock_client.close_worktree.assert_called_once_with("wt-uuid", force=True)