Add a bring-up script and a transcript dump; let nisse-dev install on its own VM (M0.2)
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m16s

Development Plan M0.2.

- scripts/bringup.sh: start [REF] sets the previous run aside under
  /root/nisse-e2e/runs/, exports the ref under test and starts the agent
  as the transient systemd unit nisse-bringup, so it outlives the shell
  that started it; stop (SIGTERM, clean shutdown), status, logs, and
  collect (the journal log and a readable transcript).
- nisse_dev.dump (python -m nisse_dev.dump STATE_DB [--full]): a readable
  dump of a state store's state, pending intents and canonical log. Opens
  the store read-only, so it is safe against a live agent.
- The agent runs as root on this development VM, as it will on its own
  VM. Architecture D3/D6/D7 corrected to match the operator's intent:
  installing packages and toolchains on its own VM is class 2, recorded
  so the environment can be rebuilt (it was class 3 with provisioning an
  open question); egress names OS package mirrors. The plan's M0.2, M4.1,
  M4.2 and bring-up guards follow.

Acceptance: a smoke run started, answered status in the test room,
stopped cleanly, and collect produced the log and transcript.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-10-01 16:16:32 +00:00
1 parent 678466c176
commit a18321b4b0
5 files changed
+288 -23

No files matched your search

+96
View File
@@ -0,0 +1,96 @@
"""A readable dump of a state store, for bring-up runs and debugging.
python -m nisse_dev.dump STATE_DB [--full]
Opens the store read-only (SQLite `mode=ro`), so it is safe to run against a
live agent's database: it never migrates, writes or locks it. Prints the
agent's state, pending intents, and the canonical log, one event per line;
long texts are shortened unless `--full` is given.
"""
import argparse
import json
import sqlite3
import sys
from pathlib import Path
from typing import Any
SHORT = 160 # characters of each text shown without --full
def _connect(path: Path) -> sqlite3.Connection:
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
return conn
def _clip(text: str, full: bool) -> str:
text = text.replace("\n", " ⏎ ") if not full else text
if full or len(text) <= SHORT:
return text
return text[:SHORT] + f" … [{len(text) - SHORT} more]"
def _event_line(kind: str, payload: dict[str, Any], full: bool) -> str:
match kind:
case "user_message":
return f"USER [{payload.get('source', '')}]: {_clip(payload['text'], full)}"
case "assistant_text":
return f"ASSISTANT ({payload.get('model', '')}): {_clip(payload['text'], full)}"
case "tool_call":
args = json.dumps(payload.get("arguments", {}), ensure_ascii=False)
return f"CALL {payload['tool']} {_clip(args, full)}"
case "tool_result":
mark = "ok" if payload.get("success") else "FAILED"
return f" result ({mark}): {_clip(str(payload.get('result', '')), full)}"
case "plan_submitted":
plan = payload["plan"]
return f"PLAN ({payload.get('model', '')}, {len(plan)} chars): {_clip(plan, full)}"
case "gate_opened":
return f"GATE {payload.get('gate', '')}: {payload.get('reason', '')}"
case _:
return f"{kind.upper()}: {_clip(json.dumps(payload, ensure_ascii=False), full)}"
def render(path: Path, *, full: bool = False) -> str:
conn = _connect(path)
lines: list[str] = []
state = conn.execute("SELECT * FROM agent_state WHERE id = 1").fetchone()
if state is None:
lines.append("state: (none)")
else:
data = json.loads(state["data"])
gate = data.get("gate") or {}
lines.append(
f"state: {state['state']} project: {state['project']} "
f"task: {data.get('task_id')} gate: "
+ (f"{gate.get('name')} rev-{gate.get('rev')}" if gate else "none")
)
pending = conn.execute(
"SELECT id, kind, payload FROM intent_log WHERE status = 'pending'"
).fetchall()
for row in pending:
lines.append(f"pending intent #{row['id']} {row['kind']}: {row['payload']}")
lines.append("")
for row in conn.execute("SELECT * FROM canonical_events ORDER BY id"):
payload = json.loads(row["payload"])
time = row["ts"][11:19]
lines.append(f"#{row['id']} {time} {_event_line(row['type'], payload, full)}")
conn.close()
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("state_db", type=Path)
parser.add_argument("--full", action="store_true", help="don't shorten texts")
args = parser.parse_args(argv)
if not args.state_db.exists():
print(f"no state store at {args.state_db}", file=sys.stderr)
return 1
print(render(args.state_db, full=args.full))
return 0
if __name__ == "__main__":
sys.exit(main())