feat(kernel): canonical intermediate-state FSM, unknown-state validation, writer timestamps

loop/contract.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import Any
 
+from . import fsm
 from .completion import (
     CompletionPolicyError,
     criteria_satisfy_completion,
@@ -218,6 +219,26 @@ def _check_tasks_semantics(data: dict[str, Any] | None, path: Path, issues: list
             issues.append(ContractIssue("done_without_evidence", f"task {task_id!r} done without evidence", path))
 
 
+def _check_state_vocabulary(
+    state: dict[str, Any] | None,
+    manifest: dict[str, Any] | None,
+    path: Path,
+    issues: list[dict],
+) -> None:
+    """Reject undeclared state names in both validation modes."""
+    if not isinstance(state, dict):
+        return
+    extra_states = manifest.get("extra_states") if isinstance(manifest, dict) else None
+    declared = (
+        tuple(value for value in extra_states if isinstance(value, str))
+        if isinstance(extra_states, list)
+        else ()
+    )
+    value = state.get("state")
+    if value not in fsm.ALL_STATES + declared:
+        issues.append(ContractIssue("unknown_state", f"unknown state {value!r}", path))
+
+
 def _validate_tasks(data: dict[str, Any] | None, path: Path, issues: list[dict]) -> None:
     _require_schema(data, "loop-engineer/tasks@1", path, issues)
     if data is None:
@@ -670,11 +691,13 @@ def validate_contract(target: str | Path) -> dict[str, Any]:
             if data is not None:
                 _jsonschema_validate(data, name, path, issues)
         # Cross-field rules JSON Schema cannot express, run in both modes.
+        _check_state_vocabulary(state, manifest, paths.state, issues)
         _check_tasks_semantics(tasks, paths.tasks, issues)
         _check_terminal_contradiction(terminal, paths.terminal, issues)
     else:
         _validate_manifest(manifest, paths.manifest, issues)
         _validate_state(state, paths.state, issues)
+        _check_state_vocabulary(state, manifest, paths.state, issues)
         _validate_tasks(tasks, paths.tasks, issues)
         _validate_terminal(terminal, paths.terminal, issues)
 

loop/emit.py

@@ -14,6 +14,7 @@
 from pathlib import Path
 from typing import Any, Sequence
 
+from . import fsm
 from .completion import (
     CompletionPolicyError,
     criteria_satisfy_completion,
@@ -127,6 +128,7 @@ def append_iteration(
     *,
     iteration_id: int,
     outcome: str,
+    state: str | None = None,
     task_id: str = "",
     actions: Sequence[str] = (),
     verify_cmd: str = "",
@@ -138,8 +140,16 @@ def append_iteration(
     .loop/state.json's iteration_id/active_task."""
     if outcome not in _ITERATION_OUTCOMES:
         raise EmitError(f"unknown iteration outcome {outcome!r}; expected one of {_ITERATION_OUTCOMES}")
+    if state is not None and (not isinstance(state, str) or not state.strip()):
+        raise EmitError("state must be a non-empty string when provided")
     _require_iteration_id(iteration_id)
     paths = _require_contract(target)
+    current = _read_state(paths)
+    if state is not None:
+        if not fsm.is_legal_transition(current.get("state"), state):
+            raise EmitError(f"illegal FSM transition {current.get('state')!r} -> {state!r}")
+        current["state"] = state
+    current["updated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
 
     lines = [
         "",
@@ -176,11 +186,10 @@ def append_iteration(
     with runlog.open("a", encoding="utf-8") as fh:
         fh.write("\n".join(lines))
 
-    state = _read_state(paths)
-    state["iteration_id"] = iteration_id
+    current["iteration_id"] = iteration_id
     if task_id:
-        state["active_task"] = task_id
-    _write_state(paths, state)
+        current["active_task"] = task_id
+    _write_state(paths, current)
     return runlog
 
 
@@ -324,7 +333,14 @@ def terminate(
         ) from exc
     except OSError as exc:
         raise EmitError(f"terminal write failed at {terminal_path}: {exc}") from exc
+    if not fsm.is_legal_transition(current.get("state"), fsm.TERMINAL_MARKER):
+        raise EmitError(
+            f"terminal written at {terminal_path} but state.json has no legal transition "
+            f"from {current.get('state')!r} to {fsm.TERMINAL_MARKER!r}"
+        )
+    current["state"] = fsm.TERMINAL_MARKER
     current["terminal_state"] = state
+    current["updated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
     try:
         _write_state(paths, current)
     except OSError as exc:
@@ -336,7 +352,7 @@ def terminate(
 
 
 def sync_state_to_terminal(target: str | Path) -> Path:
-    """Stamp state.json's ``terminal_state`` from an existing terminal record.
+    """Reconcile state.json's FSM marker and terminal verdict from its end record.
 
     The narrow repair for a crash or failed write between the immutable
     ``terminal_state.json`` creation and the state.json stamp — the two files
@@ -354,7 +370,14 @@ def sync_state_to_terminal(target: str | Path) -> Path:
     if not isinstance(terminal, dict) or terminal.get("state") not in TERMINAL_STATES:
         raise EmitError(f"terminal_state.json at {terminal_path} does not hold a valid terminal record")
     current = _read_state(paths)
+    changed = False
     if current.get("terminal_state") != terminal["state"]:
         current["terminal_state"] = terminal["state"]
+        changed = True
+    if current.get("state") != fsm.TERMINAL_MARKER:
+        current["state"] = fsm.TERMINAL_MARKER
+        changed = True
+    if changed:
+        current["updated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
         _write_state(paths, current)
     return paths.state

loop/fsm.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+
+NON_TERMINAL_STATES = (
+    "intake",
+    "plan",
+    "critique-plan",
+    "queue-tasks",
+    "execute-task",
+    "verify",
+    "repair",
+    "replan",
+    "approval-wait",
+)
+TERMINAL_MARKER = "terminal"
+ALL_STATES = NON_TERMINAL_STATES + (TERMINAL_MARKER,)
+
+
+_ACTIVE_EDGES = (
+    ("intake", "plan"),
+    ("plan", "critique-plan"),
+    ("critique-plan", "queue-tasks"),
+    ("queue-tasks", "execute-task"),
+    ("execute-task", "verify"),
+    ("verify", "execute-task"),
+    ("verify", "repair"),
+    ("repair", "verify"),
+    ("repair", "replan"),
+    ("replan", "queue-tasks"),
+)
+_APPROVAL_RESUME_TARGETS = (
+    "plan",
+    "critique-plan",
+    "queue-tasks",
+    "execute-task",
+    "verify",
+    "repair",
+    "replan",
+)
+
+
+def legal_targets(state: object) -> tuple[str, ...]:
+    """Return canonical state-changing targets in canonical vocabulary order."""
+    if state not in ALL_STATES or state == TERMINAL_MARKER:
+        return ()
+
+    def is_target(candidate: str) -> bool:
+        if candidate == state:
+            return False
+        if candidate == TERMINAL_MARKER:
+            return True
+        if candidate == "approval-wait" and state != "intake":
+            return True
+        if state == "approval-wait" and candidate in _APPROVAL_RESUME_TARGETS:
+            return True
+        return (state, candidate) in _ACTIVE_EDGES
+
+    return tuple(candidate for candidate in ALL_STATES if is_target(candidate))
+
+
+def is_legal_transition(old: object, new: object) -> bool:
+    """Check known-state adjacency; unknown endpoints deliberately fail open."""
+    if old not in ALL_STATES or new not in ALL_STATES:
+        return True
+    if old == new:
+        return True
+    return new in legal_targets(old)

reference/repo-os-contract.md

@@ -238,7 +238,8 @@ Python-FSM realization is chosen, implement the ~100-line pattern or reuse the a
 | Field | Type | Meaning |
 |---|---|---|
 | `iteration_id` | int | Monotonic loop counter (matches latest `RUNLOG` entry). |
-| `state` | enum | Current FSM state (`intake` … `verify` … `terminal`). |
+| `state` | enum | Current FSM state: `intake`, `plan`, `critique-plan`, `queue-tasks`, `execute-task`, `verify`, `repair`, `replan`, `approval-wait`, or `terminal`. `loop/fsm.py` is normative for the transition table. |
+| `updated_at` | string\|null | ISO-8601 UTC timestamp of the last write by a `loop.emit` writer; additive/optional and absent on legacy artifacts. |
 | `plan_version` | int | Bumped on every replan (lets traces detect churn). |
 | `active_task` | string\|null | `TASKS.json` id currently in flight. |
 | `best_score` | number\|null | Best verification score so far (repair productivity is measured against this). |
@@ -295,6 +296,7 @@ sets it; resolution clears it; the loop never spawns a fresh untracked attempt (
 |---|---|---|
 | `state` | enum | One of the 7 above. |
 | `iteration_id` | int | Final iteration count. |
+| `terminated_at` | string | ISO-8601 UTC timestamp stamped by `loop.emit.terminate()`; additive/optional, so legacy records without it remain valid. |
 | `criteria_met` | object | `{ "<criterion#>": true\|false }` for every `SPEC.md` criterion. |
 | `completion_policy` | object | Completion rule for the criteria map. v1 supports `{ "mode": "all_required" }`; legacy records without the field are interpreted the same way. Optional (additive). Note: a pre-migration `Succeeded` record whose criteria map contains any `false` value fails this rule and needs re-verification. |
 | `evidence` | string[] | Paths to the verification bundles backing the verdict. |

schemas/manifest.schema.json

@@ -12,6 +12,7 @@
     "outputs": { "type": "object" },
     "permissions": { "type": "array", "items": { "type": ["string", "object"] } },
     "approval_gates": { "type": "array", "items": { "type": "string" } },
+    "extra_states": { "type": "array", "items": { "type": "string", "minLength": 1 } },
     "policies": {
       "type": "object",
       "properties": {

schemas/state.schema.json

@@ -30,6 +30,13 @@
     "state": {
       "type": "string"
     },
+    "updated_at": {
+      "type": [
+        "string",
+        "null"
+      ],
+      "description": "ISO-8601 UTC timestamp of the last write by a loop.emit writer. Additive/optional; absent on legacy artifacts."
+    },
     "plan_version": {
       "type": "integer",
       "minimum": 0

schemas/terminal.schema.json

@@ -30,6 +30,10 @@
       "type": "integer",
       "minimum": 0
     },
+    "terminated_at": {
+      "type": "string",
+      "description": "ISO-8601 UTC timestamp stamped by loop.emit.terminate(). Additive/optional; legacy records without it remain valid."
+    },
     "criteria_met": {
       "type": "object",
       "additionalProperties": {

scripts/test_conformance.py

@@ -51,7 +51,7 @@
     "ITERATION_ID": "0",  # quoted in the template -> string "0" -> lifecycle "planned"
     "PLAN_VERSION": "0",
     "ACTIVE_TASK_ID": "T1",
-    "STATE": "Planned",
+    "STATE": "intake",
     "BEST_SCORE": "null",
     "FAILURE_MODE": "",
     "PENDING_APPROVAL": "null",

scripts/test_emit.py

@@ -7,6 +7,7 @@
 import json
 import threading
 from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import pytest
@@ -56,6 +57,38 @@ def test_append_iteration_rejects_unknown_outcome(workspace):
         emit.append_iteration(workspace, iteration_id=1, outcome="totally_done")
 
 
+def test_append_iteration_advances_fsm_state_when_provided(workspace):
+    emit.append_iteration(workspace, iteration_id=1, outcome="task_passed", state="plan")
+    state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8"))
+    assert state["state"] == "plan"
+
+
+def test_append_iteration_rejects_illegal_fsm_transition(workspace):
+    runlog = workspace / "RUNLOG.md"
+    before = runlog.read_text(encoding="utf-8")
+    with pytest.raises(emit.EmitError):
+        emit.append_iteration(workspace, iteration_id=1, outcome="task_passed", state="verify")
+    assert runlog.read_text(encoding="utf-8") == before
+
+
+def test_append_iteration_state_defaults_to_no_op(workspace):
+    emit.append_iteration(workspace, iteration_id=1, outcome="task_passed")
+    state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8"))
+    assert state["state"] == "intake"
+
+
+def test_append_iteration_stamps_updated_at(workspace):
+    emit.append_iteration(workspace, iteration_id=1, outcome="task_passed")
+    state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8"))
+    timestamp = datetime.fromisoformat(state["updated_at"])
+    assert timestamp.utcoffset() == timedelta(0)
+
+
+def test_append_iteration_rejects_blank_state(workspace):
+    with pytest.raises(emit.EmitError):
+        emit.append_iteration(workspace, iteration_id=1, outcome="task_passed", state="   ")
+
+
 def test_append_receipt_is_schema_valid(workspace):
     path = emit.append_receipt(
         workspace, iteration_id=1, role="write", model="claude-opus", outcome="ok"
@@ -86,6 +119,15 @@ def test_terminate_succeeded_with_evidence_passes_doctor(workspace):
     assert validate_contract(workspace)["ok"] is True
 
 
+def test_terminate_sets_state_field_to_terminal(workspace):
+    emit.terminate(
+        workspace, state="FailedBlocked", criteria_met={"1": False}, evidence=[]
+    )
+    state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8"))
+    assert state["state"] == "terminal"
+    assert datetime.fromisoformat(state["updated_at"]).utcoffset() == timedelta(0)
+
+
 @pytest.mark.parametrize(
     "kwargs",
     [
@@ -280,6 +322,23 @@ def test_sync_state_to_terminal_reconciles_unstamped_state(workspace):
     assert not _loop_leftovers(workspace)
 
 
+def test_sync_state_to_terminal_also_reconciles_state_field(workspace):
+    emit.terminate(
+        workspace, state="FailedBlocked", criteria_met={"1": False}, evidence=[]
+    )
+    state_path = workspace / ".loop" / "state.json"
+    current = json.loads(state_path.read_text(encoding="utf-8"))
+    current["state"] = "intake"
+    current.pop("updated_at")
+    state_path.write_text(json.dumps(current, indent=2) + "\n", encoding="utf-8")
+
+    emit.sync_state_to_terminal(workspace)
+
+    synced = json.loads(state_path.read_text(encoding="utf-8"))
+    assert synced["state"] == "terminal"
+    assert datetime.fromisoformat(synced["updated_at"]).utcoffset() == timedelta(0)
+
+
 def test_sync_state_to_terminal_requires_a_terminal_record(workspace):
     with pytest.raises(emit.EmitError, match="nothing to sync"):
         emit.sync_state_to_terminal(workspace)