scriptoriumadmin/smriti
1
0
Fork 0
mirror of https://github.com/himanshudongre/smriti.git synced 2026-10-08 03:08:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add smriti quickstart and a shipped demo space

Browse source 
A fresh install opens to an empty space, so Smriti's value — which only
shows once reasoning has accumulated — is invisible on day one. `smriti
quickstart` seeds one small, finished project (a rate-limiting feature
built by two agents, with a branch explored and dropped) and prints a
short guided walkthrough.

The demo space `smriti-demo` carries a marker in its description: --remove
only deletes a marked space, --reset rebuilds it, and seeding rolls back a
half-built space on failure. The fixture is plain structured data with
content-integrity tests guarding counts, intent types, note kinds, the
demo marker, and branch divergence.
This commit is contained in:
Himanshu Dongre 2026-05-17 10:49:52 +05:30
parent f4062da76e
commit 1afca2bda2
4 changed files with 1061 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cli/README.md
View file

@ -136,6 +136,7 @@ The installer is version-aware: it refuses to overwrite a destination whose inst
```
smriti init <space> [--description "..."] # one-step agent onboarding
smriti doctor # backend/runtime diagnostics
smriti quickstart [--remove | --reset] # seed a demo space + guided walkthrough
smriti space list
smriti space create <name> [--description "..."]

155
cli/smriti_cli/main.py
View file

@ -1060,6 +1060,136 @@ def cmd_init(client: SmritiClient, args: argparse.Namespace) -> None:
print()
# ── quickstart subcommand handler ───────────────────────────────────────────
def _render_quickstart(args: argparse.Namespace, payload: dict) -> None:
"""Emit a quickstart result — JSON when --json, otherwise human-readable."""
if args.json:
_print_json(payload)
return
space = payload["space"]
action = payload["action"]
print()
if action == "removed":
print(f' ✓ Removed the demo space "{space}".')
print()
return
if action == "nothing-to-remove":
print(f' No demo space to remove — "{space}" does not exist.')
print()
return
if action == "exists":
print(f' The demo space "{space}" is already seeded.')
print()
print(f" smriti current {space} explore it")
print(" smriti quickstart --reset rebuild it from scratch")
print(" smriti quickstart --remove delete it")
print()
return
# action == "seeded"
claims = payload["claims"]
active = sum(1 for c in claims if c.get("status") != "done")
n_checkpoints = len(payload["checkpoints"])
print(f' ✓ Seeded the demo space "{space}".')
print(
f" {n_checkpoints} checkpoints · 2 agents · "
f"1 branch explored and dropped · "
f"{len(claims)} work claims ({active} still active)"
)
print()
print(' This is one small, finished feature — "add rate limiting to the')
print(' API" — captured the way Smriti captures reasoning: the decisions,')
print(" the assumptions under them, a branch that was tried and dropped,")
print(" and a hand-off between two agents. Smriti's value is this")
print(" accumulated state — quickstart just gives you some on day one.")
print()
print(" Walk through it — about three minutes:")
print()
for i, step in enumerate(payload["guide"], 1):
print(f" {i}. {step['command']}")
print(f" {step['note']}")
print()
print(" Then open the dashboard: http://localhost:5173")
print()
print(" Done exploring? smriti quickstart --remove")
print()
def cmd_quickstart(client: SmritiClient, args: argparse.Namespace) -> None:
"""Seed a curated demo space so a new user sees what Smriti is for.
The empty-room problem: a fresh install opens to an empty space, and
Smriti's value only shows once reasoning has accumulated. `quickstart`
seeds one small, finished, realistic project — built by two agents, with
a branch that was explored and dropped — and prints a short walkthrough.
Default: seed `smriti-demo`. --remove deletes it; --reset removes then
re-seeds. Idempotent — re-running without flags when the demo already
exists just reprints how to explore or rebuild it.
"""
from . import quickstart as qs
# Backend reachability — same failure guidance as `smriti init`.
try:
client.list_spaces()
except SmritiError:
_fail(
f"error: Cannot reach Smriti backend at {client.base_url}.\n"
"Start the backend with `make dev-local` for solo/local mode, "
"or `make dev-postgres` for Postgres/shared-team mode."
)
return
# Removal path — both --remove and --reset clear an existing demo space.
if args.remove or args.reset:
existing = qs.find_demo_space(client)
if existing is not None and qs.is_demo_space(existing):
if not _confirm(
f'Delete the demo space "{qs.DEMO_SPACE_NAME}" and all '
f"its checkpoints?",
args.yes,
):
_fail("Cancelled.", code=0)
result = qs.remove_demo_space(client)
if not result["removed"] and result.get("reason") == "not-a-demo-space":
_fail(
f'error: A space named "{qs.DEMO_SPACE_NAME}" exists but was '
f"not created by quickstart — it lacks the demo marker.\n"
f"Refusing to delete it. To remove it yourself, run:\n"
f" smriti space delete {qs.DEMO_SPACE_NAME}"
)
return
if args.remove:
action = "removed" if result["removed"] else "nothing-to-remove"
_render_quickstart(args, {"action": action, "space": qs.DEMO_SPACE_NAME})
return
# --reset: fall through and re-seed.
# Seeding path.
if qs.find_demo_space(client) is not None:
_render_quickstart(args, {"action": "exists", "space": qs.DEMO_SPACE_NAME})
return
seed = qs.seed_demo_space(client)
_render_quickstart(
args,
{
"action": "seeded",
"space": qs.DEMO_SPACE_NAME,
"space_id": seed["space_id"],
"checkpoints": seed["checkpoints"],
"claims": seed["claims"],
"guide": qs.build_guide(seed),
},
)
# ── branch subcommand handlers ──────────────────────────────────────────────
@ -1295,6 +1425,31 @@ def _build_parser() -> argparse.ArgumentParser:
doctor_parser.add_argument("--json", action="store_true", help="Output structured JSON")
doctor_parser.set_defaults(func=cmd_doctor)
# quickstart — seed a curated demo space so the product clicks fast
quickstart_parser = subparsers.add_parser(
"quickstart",
help="Seed a curated demo space (smriti-demo) and print a guided walkthrough",
)
quickstart_mode = quickstart_parser.add_mutually_exclusive_group()
quickstart_mode.add_argument(
"--remove",
action="store_true",
help="Delete the demo space instead of seeding it",
)
quickstart_mode.add_argument(
"--reset",
action="store_true",
help="Delete the demo space if present, then seed a fresh one",
)
quickstart_parser.add_argument(
"-y",
"--yes",
action="store_true",
help="Skip the confirmation prompt when removing the demo space",
)
quickstart_parser.add_argument("--json", action="store_true")
quickstart_parser.set_defaults(func=cmd_quickstart)
# space
space_parser = subparsers.add_parser("space", help="Manage Smriti spaces (projects)")
space_sub = space_parser.add_subparsers(dest="subcommand", required=True)

515
cli/smriti_cli/quickstart.py Normal file
View file

@ -0,0 +1,515 @@
"""`smriti quickstart` — seed a curated demo space so the product clicks fast.
The empty-room problem: a fresh Smriti install opens to an empty space.
Smriti's value is emergent — it accrues as you checkpoint reasoning — so on
day one there is nothing to look at and nothing to explain why it matters.
`smriti quickstart` skips the wait. It seeds one small, finished, realistic
project — a rate-limiting feature built by two agents — captured the way
Smriti captures reasoning: decisions, the assumptions under them, a branch
that was explored and dropped, and a hand-off between agents. A new user can
then read a real continuation brief, diff a decision that diverged, and look
at coordination metrics in a few minutes instead of a few weeks.
The seeded space is named ``smriti-demo`` and is clearly marked as demo data
in its description, so it is never mistaken for the user's own work. Remove
it any time with ``smriti quickstart --remove``.
This module holds the fixture as plain structured data plus the seed/remove
logic. ``main.py`` keeps only a thin ``cmd_quickstart`` handler. The fixture
is intentionally a reduced, sanitized narrative rather than real dogfooding
history — small enough to read and maintain, representative enough to teach.
"""
from __future__ import annotations
from .client import SmritiClient, SmritiError
# ── Identity ────────────────────────────────────────────────────────────────
DEMO_SPACE_NAME = "smriti-demo"
# A sentinel embedded in the space description. `--remove` only deletes a
# space that carries this marker, so it can never delete a real space that
# happens to share the name.
DEMO_MARKER = "seeded by `smriti quickstart`"
DEMO_SPACE_DESCRIPTION = (
f"[DEMO] Sample Smriti project — a small rate-limiting feature, "
f"{DEMO_MARKER}. Safe to delete: smriti quickstart --remove"
)
DEMO_OBJECTIVE = "Add per-client rate limiting to the public API before the launch."
# ── Narrative building blocks ───────────────────────────────────────────────
#
# Decisions and assumptions accumulate down the checkpoint chain, the way real
# Smriti history does: each checkpoint carries the standing set forward and
# adds to it. Naming them once keeps the carry-forward honest and lets the
# divergence between `main` and the explored branch be exact.
_DEC_TOKEN_BUCKET = (
"Use a token-bucket limiter — it absorbs our burst shape with far less "
"per-request bookkeeping than a sliding-window log"
)
_DEC_KEY_BY_API_KEY = (
"Key the limit on the API key, not the client IP — IPs are shared behind "
"carrier NAT, so IP-based limits punish unrelated users"
)
_DEC_429_RETRY_AFTER = (
"Reject over-limit requests with HTTP 429 plus a Retry-After header, so "
"clients back off precisely instead of retrying blind"
)
_DEC_FEATURE_FLAG = (
"Ship behind the rate_limit_enabled flag — keep it off in production "
"until the load test passes"
)
_DEC_MONOTONIC_CLOCK = (
"Compute token refill from a monotonic clock, not the wall clock — an NTP "
"step was double-crediting buckets and briefly doubling a client's limit"
)
_DEC_REDIS_BUCKETS = (
"Hold the token buckets in Redis so limits stay correct across multiple "
"API instances"
)
_ASSUME_BURST = "Peak traffic is bursty but stays under ~800 requests/sec"
_ASSUME_SINGLE_INSTANCE = "The API runs as a single instance for now"
_ASSUME_MULTI_INSTANCE = "The API will scale to multiple instances within two quarters"
def _task(task_id: str, text: str, intent_type: str, status: str) -> dict:
"""A structured task entry — the shape the extractor and `state` expect."""
return {"id": task_id, "text": text, "intent_type": intent_type, "status": status}
# ── Main-branch checkpoints ─────────────────────────────────────────────────
#
# Four checkpoints, oldest first. They share one session and chain on `main`.
# `key` is an internal handle used to attach notes and to fork from a specific
# checkpoint — it is not sent to the backend.
MAIN_CHECKPOINTS: list[dict] = [
{
"key": "frame",
"author_agent": "claude-code",
"message": "Frame the rate-limiting work",
"objective": DEMO_OBJECTIVE,
"summary": (
"The public API has no abuse protection: a single client can "
"exhaust capacity for everyone. Surveyed the options. Two "
"questions block any code — which algorithm, and what the limit "
"is keyed on. No decisions yet; this checkpoint just frames the "
"work."
),
"decisions": [],
"assumptions": [_ASSUME_BURST, _ASSUME_SINGLE_INSTANCE],
"open_questions": [
"Token bucket, or a sliding-window log?",
"Do we limit per client IP, or per API key?",
],
"tasks": [
_task("survey", "Survey rate-limiting algorithms and pick candidates", "explore", "open"),
_task("decide", "Decide the algorithm and what the limit is keyed on", "decide", "open"),
_task("middleware", "Build the rate-limit middleware", "implement", "open"),
],
"entities": ["public API", "API gateway", "rate limiter"],
"artifacts": [],
},
{
"key": "decide",
"author_agent": "claude-code",
"message": "Decide: token bucket, keyed per API key",
"objective": DEMO_OBJECTIVE,
"summary": (
"Settled both open questions. Token bucket beats a sliding-window "
"log here — it absorbs our burst shape with a fraction of the "
"bookkeeping. The limit is keyed on the API key, not the client "
"IP: mobile traffic shares IPs behind carrier NAT, so IP limits "
"would punish unrelated users. Next: build the middleware."
),
"decisions": [_DEC_TOKEN_BUCKET, _DEC_KEY_BY_API_KEY],
"assumptions": [_ASSUME_BURST, _ASSUME_SINGLE_INSTANCE],
"open_questions": [],
"tasks": [
_task("survey", "Survey rate-limiting algorithms and pick candidates", "explore", "done"),
_task("decide", "Decide the algorithm and what the limit is keyed on", "decide", "done"),
_task("middleware", "Build the rate-limit middleware", "implement", "open"),
_task("loadtest", "Load-test burst and sustained traffic", "test", "open"),
],
"entities": ["public API", "API gateway", "rate limiter", "API key", "token bucket"],
"artifacts": [
{
"name": "Rate limit tiers",
"content": (
"Free tier: 60 req/min, burst 120.\n"
"Pro tier: 600 req/min, burst 1200.\n"
"Burst window: 10s. Limits are per API key."
),
}
],
},
{
"key": "ship",
"author_agent": "claude-code",
"message": "Ship the limiter middleware behind a flag",
"objective": DEMO_OBJECTIVE,
"summary": (
"Token-bucket middleware is in. Over-limit requests get HTTP 429 "
"with a Retry-After header, so well-behaved clients back off "
"instead of retrying blind. Shipped behind the rate_limit_enabled "
"flag — it stays off in production until the load test clears it."
),
"decisions": [
_DEC_TOKEN_BUCKET,
_DEC_KEY_BY_API_KEY,
_DEC_429_RETRY_AFTER,
_DEC_FEATURE_FLAG,
],
"assumptions": [_ASSUME_BURST, _ASSUME_SINGLE_INSTANCE],
"open_questions": [],
"tasks": [
_task("survey", "Survey rate-limiting algorithms and pick candidates", "explore", "done"),
_task("decide", "Decide the algorithm and what the limit is keyed on", "decide", "done"),
_task("middleware", "Build the rate-limit middleware", "implement", "done"),
_task("loadtest", "Load-test burst and sustained traffic", "test", "open"),
],
"entities": [
"public API",
"rate limiter",
"token bucket",
"429 response",
"Retry-After header",
"rate_limit_enabled flag",
],
"artifacts": [
{
"name": "429 response contract",
"content": (
"HTTP 429 Too Many Requests\n"
"Retry-After: <seconds>\n"
'Body: {"error": "rate_limited", '
'"retry_after_seconds": <n>, "limit": <n>}'
),
}
],
},
{
"key": "loadtest",
"author_agent": "codex-local",
"message": "Load test passes; fix a clock-skew refill bug",
"objective": DEMO_OBJECTIVE,
"summary": (
"Picked up claude-code's middleware and ran it under load. Held "
"750 req/sec sustained with bursts to 1500 — no false rejections. "
"Found one real bug: token refill read the wall clock, so an NTP "
"correction let a bucket refill twice and briefly doubled a "
"client's allowance. Switched refill to a monotonic clock. The "
"flag can go on."
),
"decisions": [
_DEC_TOKEN_BUCKET,
_DEC_KEY_BY_API_KEY,
_DEC_429_RETRY_AFTER,
_DEC_FEATURE_FLAG,
_DEC_MONOTONIC_CLOCK,
],
"assumptions": [_ASSUME_BURST, _ASSUME_SINGLE_INSTANCE],
"open_questions": [],
"tasks": [
_task("survey", "Survey rate-limiting algorithms and pick candidates", "explore", "done"),
_task("decide", "Decide the algorithm and what the limit is keyed on", "decide", "done"),
_task("middleware", "Build the rate-limit middleware", "implement", "done"),
_task("loadtest", "Load-test burst and sustained traffic", "test", "done"),
_task("document", "Document the 429 / Retry-After contract for API consumers", "docs", "open"),
],
"entities": [
"public API",
"rate limiter",
"token bucket",
"monotonic clock",
"load test",
],
"artifacts": [],
},
]
# ── The branch that was explored and dropped ────────────────────────────────
#
# Forked from the `decide` checkpoint. It revises the single-instance
# assumption rather than carrying it forward — that revision is the exact
# point of divergence `smriti compare` surfaces against `main`.
FORK_CHECKPOINT: dict = {
"key": "redis-spike",
"fork_from": "decide",
"branch_name": "explore/redis-limiter",
"disposition": "abandoned",
"author_agent": "codex-local",
"message": "Spike: a Redis-backed distributed limiter",
"objective": DEMO_OBJECTIVE,
"summary": (
"Explored keeping the token buckets in Redis so limits stay correct "
"if we ever run more than one API instance. The spike works, but it "
"adds a hard Redis dependency and ~3ms of latency to every request. "
"We run a single instance today, so this buys correctness we do not "
"need yet. Parking it — revisit if we scale out."
),
"decisions": [_DEC_TOKEN_BUCKET, _DEC_KEY_BY_API_KEY, _DEC_REDIS_BUCKETS],
"assumptions": [_ASSUME_BURST, _ASSUME_MULTI_INSTANCE],
"open_questions": ["Is ~3ms of added per-request latency acceptable at the edge?"],
"tasks": [
_task("survey", "Survey rate-limiting algorithms and pick candidates", "explore", "done"),
_task("decide", "Decide the algorithm and what the limit is keyed on", "decide", "done"),
_task("redis-spike", "Prototype the Redis token-bucket store", "explore", "done"),
],
"entities": ["rate limiter", "token bucket", "Redis", "distributed rate limiter"],
"artifacts": [],
}
# ── Notes — one of each kind ─────────────────────────────────────────────────
NOTES: list[dict] = [
{
"checkpoint_key": "ship",
"kind": "milestone",
"author": "claude-code",
"text": (
"First working rate limiter merged. 429s with Retry-After are "
"live behind the flag — the launch blocker is cleared pending "
"the load test."
),
},
{
"checkpoint_key": "decide",
"kind": "note",
"author": "founder",
"text": (
"Make the per-key limits configurable per plan tier — enterprise "
"customers will ask for higher ceilings on day one. Do not "
"hard-code these."
),
},
{
"checkpoint_key": "frame",
"kind": "noise",
"author": "codex-local",
"text": (
"Renamed the scratch branch from rl-test to explore/redis-limiter "
"so it reads cleanly in the lineage view."
),
},
]
# ── Work claims — one finished, one still open ───────────────────────────────
#
# intent_type values must be in the backend's VALID_INTENT_TYPES set
# {implement, review, investigate, docs, test}; test_quickstart.py guards this.
CLAIMS: list[dict] = [
{
"agent": "claude-code",
"scope": "Ship the token-bucket rate-limit middleware",
"intent_type": "implement",
"task_id": "middleware",
"final_status": "done",
},
{
"agent": "codex-local",
"scope": "Document the 429 / Retry-After contract for API consumers",
"intent_type": "docs",
"task_id": "document",
"final_status": "active",
},
]
# A long TTL so the demo's one active claim stays visible as "active work"
# for the lifetime of the demo space rather than expiring in a few hours.
_DEMO_CLAIM_TTL_HOURS = 720.0
# ── Seeding ─────────────────────────────────────────────────────────────────
def _commit_payload(space_id: str, session_id: str, checkpoint: dict) -> dict:
"""Build a /api/v4/chat/commit payload from a fixture checkpoint."""
return {
"repo_id": space_id,
"session_id": session_id,
"message": checkpoint["message"],
"summary": checkpoint.get("summary", ""),
"objective": checkpoint.get("objective", ""),
"decisions": checkpoint.get("decisions", []),
"assumptions": checkpoint.get("assumptions", []),
"tasks": checkpoint.get("tasks", []),
"open_questions": checkpoint.get("open_questions", []),
"entities": checkpoint.get("entities", []),
"artifacts": checkpoint.get("artifacts", []),
"author_agent": checkpoint["author_agent"],
}
def _populate(client: SmritiClient, space_id: str) -> dict:
"""Write the full demo narrative into an already-created space."""
checkpoints: dict[str, dict] = {} # fixture key -> {id, hash, branch}
# Main branch — one session; the four checkpoints chain on it.
session = client.create_session(repo_id=space_id, title="smriti quickstart demo")
main_session_id = session["id"]
for checkpoint in MAIN_CHECKPOINTS:
commit = client.create_chat_commit(
_commit_payload(space_id, main_session_id, checkpoint)
)
checkpoints[checkpoint["key"]] = {
"id": commit["id"],
"hash": commit.get("commit_hash", ""),
"branch": commit.get("branch_name", "main"),
}
# The branch that was explored, then dropped.
fork_source = checkpoints[FORK_CHECKPOINT["fork_from"]]
fork = client.fork_session(
space_id=space_id,
checkpoint_id=fork_source["id"],
branch_name=FORK_CHECKPOINT["branch_name"],
)
branch_session_id = fork["session_id"]
branch_name = fork.get("branch_name") or FORK_CHECKPOINT["branch_name"]
branch_commit = client.create_chat_commit(
_commit_payload(space_id, branch_session_id, FORK_CHECKPOINT)
)
checkpoints[FORK_CHECKPOINT["key"]] = {
"id": branch_commit["id"],
"hash": branch_commit.get("commit_hash", ""),
"branch": branch_name,
}
client.close_branch(space_id, branch_name, FORK_CHECKPOINT["disposition"])
# Notes — a milestone, a founder note, and one low-signal `noise` note.
for note in NOTES:
client.add_checkpoint_note(
checkpoint_id=checkpoints[note["checkpoint_key"]]["id"],
text=note["text"],
author=note["author"],
kind=note["kind"],
)
# Work claims — one finished, one left open so "active work" is visible.
claims: list[dict] = []
for spec in CLAIMS:
claim = client.create_claim(
space_id=space_id,
agent=spec["agent"],
scope=spec["scope"],
branch_name="main",
task_id=spec.get("task_id"),
intent_type=spec["intent_type"],
ttl_hours=_DEMO_CLAIM_TTL_HOURS,
)
if spec["final_status"] == "done":
client.update_claim(claim["id"], "done")
claims.append({"id": claim["id"], "status": spec["final_status"]})
return {
"space_id": space_id,
"checkpoints": checkpoints,
"branch_name": branch_name,
"claims": claims,
}
def seed_demo_space(client: SmritiClient) -> dict:
"""Create the `smriti-demo` space and populate it with the demo narrative.
Returns a summary dict: ``space_id``, ``checkpoints`` (fixture key ->
{id, hash, branch}), ``branch_name``, and ``claims``.
If anything fails partway through, the half-built space is removed on a
best-effort basis and the error is re-raised — so a retry always starts
from a clean slate rather than a duplicate-name conflict.
"""
space = client.create_space(
name=DEMO_SPACE_NAME, description=DEMO_SPACE_DESCRIPTION
)
space_id = space["id"]
try:
return _populate(client, space_id)
except Exception:
try:
client.delete_space(space_id)
except SmritiError:
pass # best-effort rollback; surface the original error
raise
# ── Lookup / removal ────────────────────────────────────────────────────────
def find_demo_space(client: SmritiClient) -> dict | None:
"""Return the `smriti-demo` space dict, or None if it does not exist."""
try:
return client.resolve_space(DEMO_SPACE_NAME)
except SmritiError:
return None
def is_demo_space(space: dict) -> bool:
"""True only for a space quickstart actually seeded.
Guards `--remove` from ever deleting a real space that merely shares the
`smriti-demo` name — only a space carrying the demo marker is removable.
"""
return DEMO_MARKER in (space.get("description") or "")
def remove_demo_space(client: SmritiClient) -> dict:
"""Delete the demo space if it is safe to.
Returns a result dict with ``removed`` (bool) and a ``reason``:
- ``removed=True`` — deleted
- ``reason="no-demo-space"`` — nothing named smriti-demo exists
- ``reason="not-a-demo-space"`` — a space exists but lacks the marker
"""
space = find_demo_space(client)
if space is None:
return {"removed": False, "reason": "no-demo-space"}
if not is_demo_space(space):
return {
"removed": False,
"reason": "not-a-demo-space",
"space_id": space["id"],
}
client.delete_space(space["id"])
return {"removed": True, "space_id": space["id"]}
# ── Guided walkthrough ──────────────────────────────────────────────────────
def build_guide(seed: dict) -> list[dict]:
"""Build the post-seed walkthrough — commands with real IDs filled in."""
checkpoints = seed["checkpoints"]
decide_id = checkpoints["decide"]["id"]
redis_id = checkpoints["redis-spike"]["id"]
return [
{
"command": f"smriti current {DEMO_SPACE_NAME}",
"note": "Where the project stands: direction, last milestone, what's open.",
},
{
"command": f"smriti state {DEMO_SPACE_NAME}",
"note": "The full continuation brief an agent receives at session start.",
},
{
"command": f"smriti checkpoint list {DEMO_SPACE_NAME}",
"note": "The reasoning history — five checkpoints across two agents.",
},
{
"command": f"smriti compare {decide_id} {redis_id}",
"note": "A decision that diverged: the Redis limiter explored, then dropped.",
},
{
"command": f"smriti metrics {DEMO_SPACE_NAME}",
"note": "Coordination and quality signal across the two agents.",
},
]

390
cli/tests/test_quickstart.py Normal file
View file

@ -0,0 +1,390 @@
"""Tests for `smriti quickstart` and the shipped demo-space fixture.
Covers:
- Parser wiring (flags, mutually exclusive --remove/--reset)
- Fixture integrity — counts, valid intent types / note kinds, the demo
marker, branch divergence, structured tasks. These are content-integrity
guards: the demo is shipped data, and these tests fail loudly if an edit
silently breaks the narrative it is meant to teach.
- seed_demo_space drives the expected sequence of API calls, and rolls back
a half-built space on failure
- cmd_quickstart: seeds when absent, is idempotent when present, fails
cleanly on an unreachable backend
- remove_demo_space deletes only a marked demo space; refuses a look-alike
Like the rest of cli/tests, these use MagicMock(spec=SmritiClient) — no
running backend required.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from smriti_cli import main as cli_main
from smriti_cli.client import SmritiClient, SmritiError
from smriti_cli.quickstart import (
CLAIMS,
DEMO_MARKER,
DEMO_SPACE_DESCRIPTION,
DEMO_SPACE_NAME,
FORK_CHECKPOINT,
MAIN_CHECKPOINTS,
NOTES,
build_guide,
find_demo_space,
is_demo_space,
remove_demo_space,
seed_demo_space,
)
# These mirror the backend's validation sets (claims.py VALID_INTENT_TYPES,
# checkpoint.py VALID_NOTE_KINDS). If the demo fixture ever uses a value
# outside them, seeding would 400 against a real backend — catch it here.
BACKEND_INTENT_TYPES = {"implement", "review", "investigate", "docs", "test"}
BACKEND_NOTE_KINDS = {"note", "milestone", "noise"}
# ── helpers ──────────────────────────────────────────────────────────────────
def _make_seed_client() -> MagicMock:
"""A MagicMock(spec=SmritiClient) with every seed-path method configured."""
client = MagicMock(spec=SmritiClient)
client.base_url = "http://localhost:8000"
client.list_spaces.return_value = []
client.create_space.return_value = {"id": "demo-space-id", "name": DEMO_SPACE_NAME}
client.create_session.return_value = {"id": "session-main"}
commit_counter = iter(range(1, 9999))
def _commit(payload):
n = next(commit_counter)
return {"id": f"commit-{n}", "commit_hash": f"hash{n:04d}", "branch_name": "main"}
client.create_chat_commit.side_effect = _commit
client.fork_session.return_value = {
"session_id": "session-fork",
"branch_name": "explore/redis-limiter",
}
client.close_branch.return_value = {"branch_name": "explore/redis-limiter"}
client.add_checkpoint_note.return_value = {"checkpoint_id": "cp", "kind": "note"}
claim_counter = iter(range(1, 9999))
def _claim(**kwargs):
return {"id": f"claim-{next(claim_counter)}", **kwargs}
client.create_claim.side_effect = _claim
client.update_claim.return_value = {"id": "claim-1", "status": "done"}
return client
def _args(*argv: str):
"""Parse a quickstart argv and stub api_url, the way main() would."""
parsed = cli_main._build_parser().parse_args(["quickstart", *argv])
parsed.api_url = None
return parsed
# ── parser wiring ────────────────────────────────────────────────────────────
def test_quickstart_parser_wiring():
args = _args()
assert args.command == "quickstart"
assert args.func is cli_main.cmd_quickstart
assert args.remove is False and args.reset is False
assert args.yes is False and args.json is False
assert _args("--remove").remove is True
reset = _args("--reset", "-y")
assert reset.reset is True and reset.yes is True
def test_quickstart_remove_and_reset_are_mutually_exclusive():
with pytest.raises(SystemExit):
cli_main._build_parser().parse_args(["quickstart", "--remove", "--reset"])
# ── fixture integrity ────────────────────────────────────────────────────────
def test_fixture_has_four_main_checkpoints_and_one_fork():
assert len(MAIN_CHECKPOINTS) == 4
assert FORK_CHECKPOINT["key"] == "redis-spike"
assert FORK_CHECKPOINT["disposition"] == "abandoned"
def test_fixture_uses_two_distinct_agents():
authors = {cp["author_agent"] for cp in MAIN_CHECKPOINTS}
authors.add(FORK_CHECKPOINT["author_agent"])
assert authors == {"claude-code", "codex-local"}
def test_fixture_has_a_cross_agent_continuation():
"""At least one adjacent main checkpoint pair changes author — the
hand-off that makes the coordination metric non-zero."""
pairs = zip(MAIN_CHECKPOINTS, MAIN_CHECKPOINTS[1:])
assert any(a["author_agent"] != b["author_agent"] for a, b in pairs)
def test_fixture_note_kinds_cover_all_three():
kinds = {note["kind"] for note in NOTES}
assert kinds == BACKEND_NOTE_KINDS
def test_fixture_note_kinds_are_backend_valid():
for note in NOTES:
assert note["kind"] in BACKEND_NOTE_KINDS
def test_fixture_notes_reference_real_checkpoints():
valid_keys = {cp["key"] for cp in MAIN_CHECKPOINTS} | {FORK_CHECKPOINT["key"]}
for note in NOTES:
assert note["checkpoint_key"] in valid_keys
def test_fixture_claim_intent_types_are_backend_valid():
for claim in CLAIMS:
assert claim["intent_type"] in BACKEND_INTENT_TYPES
def test_fixture_has_one_done_and_one_active_claim():
statuses = {claim["final_status"] for claim in CLAIMS}
assert statuses == {"done", "active"}
def test_fixture_demo_space_is_obviously_demo_data():
assert DEMO_SPACE_NAME == "smriti-demo"
assert DEMO_SPACE_DESCRIPTION.startswith("[DEMO]")
assert DEMO_MARKER in DEMO_SPACE_DESCRIPTION
def test_fixture_fork_source_is_a_real_main_checkpoint():
main_keys = {cp["key"] for cp in MAIN_CHECKPOINTS}
assert FORK_CHECKPOINT["fork_from"] in main_keys
def test_fixture_branch_diverges_from_its_fork_source():
"""The explored branch must disagree with its source on at least one
decision or assumption, so `smriti compare` shows a real divergence."""
source = next(
cp for cp in MAIN_CHECKPOINTS if cp["key"] == FORK_CHECKPOINT["fork_from"]
)
new_decisions = set(FORK_CHECKPOINT["decisions"]) - set(source["decisions"])
new_assumptions = set(FORK_CHECKPOINT["assumptions"]) - set(source["assumptions"])
assert new_decisions or new_assumptions
def test_fixture_tasks_are_well_formed():
for cp in [*MAIN_CHECKPOINTS, FORK_CHECKPOINT]:
for task in cp["tasks"]:
assert {"id", "text", "intent_type", "status"} <= set(task)
assert task["status"] in {"open", "done"}
def test_fixture_decisions_accumulate_down_the_main_chain():
"""Each main checkpoint carries the standing decisions forward — so the
HEAD checkpoint (and `current`/`state`) shows the full picture."""
seen = [len(cp["decisions"]) for cp in MAIN_CHECKPOINTS]
assert seen == sorted(seen)
assert seen[-1] >= 4
# ── seed_demo_space ──────────────────────────────────────────────────────────
def test_seed_demo_space_makes_expected_api_calls():
client = _make_seed_client()
seed_demo_space(client)
client.create_space.assert_called_once()
client.create_session.assert_called_once()
# 4 main checkpoints + 1 on the explored branch
assert client.create_chat_commit.call_count == 5
client.fork_session.assert_called_once()
client.close_branch.assert_called_once()
assert client.add_checkpoint_note.call_count == 3
assert client.create_claim.call_count == 2
# exactly one claim is marked done; the other stays active
client.update_claim.assert_called_once()
def test_seed_closes_the_explored_branch_as_abandoned():
client = _make_seed_client()
seed_demo_space(client)
space_id, branch_name, disposition = client.close_branch.call_args[0]
assert disposition == "abandoned"
def test_seed_marks_the_implement_claim_done():
client = _make_seed_client()
seed_demo_space(client)
assert client.update_claim.call_args[0][1] == "done"
def test_seed_uses_a_long_ttl_so_active_work_stays_visible():
client = _make_seed_client()
seed_demo_space(client)
for call in client.create_claim.call_args_list:
assert call.kwargs["ttl_hours"] >= 168 # at least a week
def test_seed_returns_handles_for_every_checkpoint():
client = _make_seed_client()
result = seed_demo_space(client)
assert result["space_id"] == "demo-space-id"
expected = {cp["key"] for cp in MAIN_CHECKPOINTS} | {FORK_CHECKPOINT["key"]}
assert set(result["checkpoints"]) == expected
assert len(result["claims"]) == 2
def test_seed_rolls_back_a_half_built_space_on_failure():
client = _make_seed_client()
client.create_chat_commit.side_effect = SmritiError("backend exploded")
with pytest.raises(SmritiError):
seed_demo_space(client)
# the partially-created space is removed so a retry starts clean
client.delete_space.assert_called_once_with("demo-space-id")
# ── build_guide ──────────────────────────────────────────────────────────────
def test_build_guide_fills_in_real_checkpoint_ids():
seed = {
"space_id": "s",
"checkpoints": {
"frame": {"id": "c-frame"},
"decide": {"id": "c-decide"},
"ship": {"id": "c-ship"},
"loadtest": {"id": "c-loadtest"},
"redis-spike": {"id": "c-redis"},
},
"branch_name": "explore/redis-limiter",
"claims": [],
}
guide = build_guide(seed)
assert len(guide) == 5
compare = next(s for s in guide if s["command"].startswith("smriti compare"))
assert "c-decide" in compare["command"]
assert "c-redis" in compare["command"]
# ── remove_demo_space ────────────────────────────────────────────────────────
def test_remove_deletes_a_marked_demo_space():
client = MagicMock(spec=SmritiClient)
client.resolve_space.return_value = {
"id": "demo-space-id",
"name": DEMO_SPACE_NAME,
"description": DEMO_SPACE_DESCRIPTION,
}
result = remove_demo_space(client)
assert result["removed"] is True
client.delete_space.assert_called_once_with("demo-space-id")
def test_remove_refuses_a_space_without_the_demo_marker():
"""A real space that merely shares the smriti-demo name must survive."""
client = MagicMock(spec=SmritiClient)
client.resolve_space.return_value = {
"id": "real-space-id",
"name": DEMO_SPACE_NAME,
"description": "My actual project. Do not delete.",
}
result = remove_demo_space(client)
assert result["removed"] is False
assert result["reason"] == "not-a-demo-space"
client.delete_space.assert_not_called()
def test_remove_when_no_demo_space_exists():
client = MagicMock(spec=SmritiClient)
client.resolve_space.side_effect = SmritiError("no space found")
result = remove_demo_space(client)
assert result["removed"] is False
assert result["reason"] == "no-demo-space"
client.delete_space.assert_not_called()
def test_is_demo_space_only_true_for_marked_spaces():
assert is_demo_space({"description": DEMO_SPACE_DESCRIPTION}) is True
assert is_demo_space({"description": "something else"}) is False
assert is_demo_space({}) is False
def test_find_demo_space_returns_none_when_absent():
client = MagicMock(spec=SmritiClient)
client.resolve_space.side_effect = SmritiError("not found")
assert find_demo_space(client) is None
# ── cmd_quickstart ───────────────────────────────────────────────────────────
def test_cmd_quickstart_seeds_when_space_absent():
client = _make_seed_client()
client.resolve_space.side_effect = SmritiError("not found")
cli_main.cmd_quickstart(client, _args())
client.create_space.assert_called_once()
def test_cmd_quickstart_is_idempotent_when_space_present(capsys):
client = MagicMock(spec=SmritiClient)
client.base_url = "http://localhost:8000"
client.list_spaces.return_value = []
client.resolve_space.return_value = {
"id": "demo-space-id",
"name": DEMO_SPACE_NAME,
"description": DEMO_SPACE_DESCRIPTION,
}
cli_main.cmd_quickstart(client, _args())
client.create_space.assert_not_called()
assert "already seeded" in capsys.readouterr().out
def test_cmd_quickstart_fails_cleanly_on_unreachable_backend(capsys):
client = MagicMock(spec=SmritiClient)
client.base_url = "http://localhost:8000"
client.list_spaces.side_effect = SmritiError("connection refused")
with pytest.raises(SystemExit):
cli_main.cmd_quickstart(client, _args())
err = capsys.readouterr().err
assert "make dev-local" in err
def test_cmd_quickstart_remove_deletes_the_demo_space():
client = MagicMock(spec=SmritiClient)
client.base_url = "http://localhost:8000"
client.list_spaces.return_value = []
client.resolve_space.return_value = {
"id": "demo-space-id",
"name": DEMO_SPACE_NAME,
"description": DEMO_SPACE_DESCRIPTION,
}
cli_main.cmd_quickstart(client, _args("--remove", "-y"))
client.delete_space.assert_called_once_with("demo-space-id")
def test_cmd_quickstart_reset_removes_then_reseeds():
client = _make_seed_client()
# present on the first lookup (removal), absent on the second (seed)
client.resolve_space.side_effect = [
{
"id": "old-demo-id",
"name": DEMO_SPACE_NAME,
"description": DEMO_SPACE_DESCRIPTION,
},
{
"id": "old-demo-id",
"name": DEMO_SPACE_NAME,
"description": DEMO_SPACE_DESCRIPTION,
},
SmritiError("not found"),
]
cli_main.cmd_quickstart(client, _args("--reset", "-y"))
client.delete_space.assert_called_once_with("old-demo-id")
client.create_space.assert_called_once()