Add local operator smoke runner

pyproject.toml

@@ -53,6 +53,7 @@ agent-term = "agent_term.cli:main"
 agent-term-check = "agent_term.health_cli:main"
 agent-term-dispatch = "agent_term.dispatch_cli:main"
 agent-term-matrix = "agent_term.matrix_cli:main"
+agent-term-smoke = "agent_term.operator_smoke_cli:main"
 agent-term-snapshot = "agent_term.snapshot_cli:main"
 
 [tool.setuptools.packages.find]

src/agent_term/operator_smoke.py

@@ -0,0 +1,171 @@
+"""Local operator smoke runner for AgentTerm.
+
+The smoke runner executes the documented local operator path without requiring live
+Matrix, Agent Registry, or Policy Fabric services. It is intentionally explicit and
+records all state into a caller-provided work directory.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from agent_term.dispatch_cli import main as dispatch_main
+from agent_term.health_cli import main as health_main
+from agent_term.matrix_cli import main as matrix_main
+from agent_term.snapshot_cli import main as snapshot_main
+
+
+@dataclass(frozen=True)
+class SmokeStep:
+    """One executed smoke step."""
+
+    name: str
+    exit_code: int
+    command: tuple[str, ...]
+
+    @property
+    def ok(self) -> bool:
+        return self.exit_code == 0
+
+
+@dataclass(frozen=True)
+class SmokeResult:
+    """Complete local operator smoke result."""
+
+    workdir: Path
+    steps: tuple[SmokeStep, ...] = field(default_factory=tuple)
+
+    @property
+    def ok(self) -> bool:
+        return all(step.ok for step in self.steps)
+
+    def render_text(self) -> str:
+        lines = [f"smoke_status={'ok' if self.ok else 'failed'}", f"workdir={self.workdir}"]
+        for step in self.steps:
+            status = "ok" if step.ok else "failed"
+            lines.append(f"{step.name}\t{status}\texit={step.exit_code}\t{' '.join(step.command)}")
+        return "\n".join(lines)
+
+
+DEFAULT_CONFIG = Path("configs/agent-term.local.example.json")
+DEFAULT_MATRIX_SYNC = Path("configs/fixtures/matrix-sync.local.example.json")
+
+
+def run_local_operator_smoke(
+    *,
+    workdir: Path,
+    config_path: Path = DEFAULT_CONFIG,
+    matrix_sync_fixture: Path = DEFAULT_MATRIX_SYNC,
+) -> SmokeResult:
+    """Run the local operator quickstart flow against isolated state."""
+
+    workdir.mkdir(parents=True, exist_ok=True)
+    db_path = workdir / "events.sqlite3"
+    state_path = workdir / "matrix-state.json"
+
+    steps = [
+        _run(
+            "check",
+            health_main,
+            (
+                "--config",
+                str(config_path),
+                "--agent-id",
+                "agent.github",
+                "--tool",
+                "repo-write",
+                "--policy-action",
+                "github.pr.create",
+            ),
+        ),
+        _run(
+            "matrix-normalize-sync",
+            matrix_main,
+            (
+                "--config",
+                str(config_path),
+                "--db",
+                str(db_path),
+                "--state",
+                str(state_path),
+                "normalize-sync",
+                str(matrix_sync_fixture),
+                "--persist",
+                "--save-state",
+            ),
+        ),
+        _run(
+            "matrix-send",
+            matrix_main,
+            (
+                "--config",
+                str(config_path),
+                "--db",
+                str(db_path),
+                "--state",
+                str(state_path),
+                "send",
+                "sourceosOps",
+                "AgentTerm local smoke runner is online.",
+            ),
+        ),
+        _run(
+            "github-dispatch",
+            dispatch_main,
+            (
+                "--config",
+                str(config_path),
+                "--db",
+                str(db_path),
+                "--tool",
+                "repo-write",
+                "--policy-action",
+                "github.pr.create",
+                "github",
+                "github_mutation",
+                "!github",
+                "Create PR for AgentTerm local smoke runner",
+            ),
+        ),
+        _run(
+            "memory-dispatch",
+            dispatch_main,
+            (
+                "--config",
+                str(config_path),
+                "--db",
+                str(db_path),
+                "--metadata-json",
+                json.dumps(
+                    {
+                        "query": "operator smoke runner context",
+                        "policy_action": "memory-mesh.memory_recall",
+                        "workroom": "operator-smoke",
+                        "topic_scope": "sourceos-agentterm",
+                    },
+                    sort_keys=True,
+                ),
+                "memory-mesh",
+                "memory_recall",
+                "!memory-mesh",
+                "Recall operator smoke runner context",
+            ),
+        ),
+        _run(
+            "snapshot",
+            snapshot_main,
+            ("--db", str(db_path), "--limit", "200"),
+        ),
+    ]
+
+    return SmokeResult(workdir=workdir, steps=tuple(steps))
+
+
+def _run(name: str, fn, argv: tuple[str, ...]) -> SmokeStep:
+    try:
+        exit_code = int(fn(list(argv)))
+    except SystemExit as exc:
+        exit_code = int(exc.code or 0)
+    return SmokeStep(name=name, exit_code=exit_code, command=argv)

src/agent_term/operator_smoke_cli.py

@@ -0,0 +1,43 @@
+"""CLI entry point for AgentTerm local operator smoke tests."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from agent_term.operator_smoke import DEFAULT_CONFIG, DEFAULT_MATRIX_SYNC, run_local_operator_smoke
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="agent-term-smoke",
+        description="Run the local AgentTerm operator smoke path using offline fixtures.",
+    )
+    parser.add_argument(
+        "--workdir",
+        default=".agent-term/smoke",
+        help="Directory for isolated smoke state.",
+    )
+    parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="AgentTerm config path.")
+    parser.add_argument(
+        "--matrix-sync-fixture",
+        default=str(DEFAULT_MATRIX_SYNC),
+        help="Matrix sync fixture path.",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    result = run_local_operator_smoke(
+        workdir=Path(args.workdir),
+        config_path=Path(args.config),
+        matrix_sync_fixture=Path(args.matrix_sync_fixture),
+    )
+    print(result.render_text())
+    return 0 if result.ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

tests/test_operator_smoke.py

@@ -0,0 +1,67 @@
+import json
+from pathlib import Path
+
+from agent_term.operator_smoke import run_local_operator_smoke
+from agent_term.operator_smoke_cli import main
+from agent_term.store import EventStore
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+CONFIG_PATH = REPO_ROOT / "configs" / "agent-term.local.example.json"
+MATRIX_SYNC_FIXTURE = REPO_ROOT / "configs" / "fixtures" / "matrix-sync.local.example.json"
+
+
+def test_local_operator_smoke_runner_executes_quickstart_flow(tmp_path):
+    result = run_local_operator_smoke(
+        workdir=tmp_path,
+        config_path=CONFIG_PATH,
+        matrix_sync_fixture=MATRIX_SYNC_FIXTURE,
+    )
+
+    assert result.ok is True
+    assert [step.name for step in result.steps] == [
+        "check",
+        "matrix-normalize-sync",
+        "matrix-send",
+        "github-dispatch",
+        "memory-dispatch",
+        "snapshot",
+    ]
+    assert (tmp_path / "events.sqlite3").exists()
+    assert (tmp_path / "matrix-state.json").exists()
+
+    state = json.loads((tmp_path / "matrix-state.json").read_text(encoding="utf-8"))
+    assert state["next_batch"] == "local-batch-2"
+
+    store = EventStore(tmp_path / "events.sqlite3")
+    try:
+        events = store.tail(limit=100)
+    finally:
+        store.close()
+
+    sources = [event.source for event in events]
+    assert "matrix" in sources
+    assert "matrix-service" in sources
+    assert "agent-registry" in sources
+    assert "policy-fabric" in sources
+    assert "github" in sources
+    assert "memory-mesh" in sources
+
+
+def test_operator_smoke_cli_reports_success(tmp_path, capsys):
+    exit_code = main(
+        [
+            "--workdir",
+            str(tmp_path),
+            "--config",
+            str(CONFIG_PATH),
+            "--matrix-sync-fixture",
+            str(MATRIX_SYNC_FIXTURE),
+        ]
+    )
+
+    captured = capsys.readouterr()
+    assert exit_code == 0
+    assert "smoke_status=ok" in captured.out
+    assert "check\tok\texit=0" in captured.out
+    assert "snapshot\tok\texit=0" in captured.out