dongarandClaude Opus 5.5
e351d0a74e
A pure move; behaviour unchanged. - models (L9, L7): llm, catalog, probe, render, render_anthropic, context - log (L7, L6): canonical, views - state (L8): store (was state), reconcile, worker, cancel, dump - knowledge (L10, L6.4): vault, memory - conversation (L5): transport, matrix - execution (L13): ssh (was execution), testing - authority (P3, P4, L4): actions - gitea, oplog, envfile stay at the top until each gains a second module. Every import, monkeypatch target, script and doc reference rewritten. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""Integration: state written before an abrupt process death must be intact
|
|
and the in-flight intent visible on the next boot (architecture L8)."""
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nisse_core.state.reconcile import reconcile_on_boot
|
|
from nisse_dev.state import StateStore
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
CRASH_SCRIPT = """
|
|
import os
|
|
import sys
|
|
from nisse_core.log.canonical import UserMessage
|
|
from nisse_dev.state import StateStore
|
|
|
|
store = StateStore(sys.argv[1])
|
|
store.open_work_item("demo-1", title="t", scope="demo", position="RUNNING", data={"milestone": 2}, cause="test")
|
|
store.set_active("demo-1", cause="test")
|
|
store.begin_intent("push_branch", {"branch": "milestone-2"}, cause="test")
|
|
store.transcript("t", "operator").append(UserMessage(text="start milestone 2", source="m:1"))
|
|
os._exit(1) # die without closing anything, as a power cut would
|
|
"""
|
|
|
|
|
|
def test_state_survives_process_death(tmp_path: Path) -> None:
|
|
db = tmp_path / "state.db"
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", CRASH_SCRIPT, str(db)],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1, result.stderr
|
|
|
|
with StateStore(db) as store:
|
|
item = store.active_item()
|
|
assert item is not None
|
|
assert item.position == "RUNNING"
|
|
assert item.data == {"milestone": 2}
|
|
|
|
report = reconcile_on_boot(store, {"push_branch": lambda intent: None})
|
|
assert report.action == "await_input"
|
|
[pending] = store.pending_intents()
|
|
assert pending.payload == {"branch": "milestone-2"}
|
|
|
|
[stored] = store.all_events()
|
|
assert stored.event.text == "start milestone 2"
|