qa: 'system' entry ingestion + integration_test.py re-baselined on the consolidated goal world — the daemon's harness-authored rows (session-restore footnote, goal park notice) ride the wire typed 'system' and author system comment rows on the ONE thread (never agent); the harness's arc now covers: commentary-only and trailing-comment rounds fed verbatim with their receipts reaching the model, the degenerate repetition validator-caught cross-binary (one rejected entry, never executed, clean resend completes), the terminal watermark advancing past the worker's transition, the derived narrative's system rows, live push chips and the accept-refuses-while-origin-is-behind gate (rewind -> refuses with the clear message, goal stays in review, re-push -> chip restores, accept passes). 83 -> 100 checks, green end-to-end against both real binaries

src/Services/SaddleService.vala

@@ -1295,6 +1295,15 @@ namespace ClankerSaddle.Services {
                             yield tasks.add_comment(task_id, "agent", body);
                         }
                         break;
+                    case "system":
+                        // Plan 15 C2: harness-authored bookkeeping (the
+                        // session-restore footnote, the goal park
+                        // notice) — the distinct system-event class on
+                        // the ONE thread, never an agent row.
+                        if (output.length > 0) {
+                            yield tasks.add_system_comment(task_id, output);
+                        }
+                        break;
                     case "plan":
                         // Plan 12: plans are goal-context only — a
                         // goal-run's `saddle plan` enters plan review;

test/integration_test.py

@@ -8,27 +8,36 @@ WORKSPACE, powered by saddle/test/mock_llm.py with a deterministic
 script — both binaries, the true cross-binary contract proof (the
 per-repo fake-peer suits live in web_stack_test.py / saddle e2e).
 
-The goal-era arc (plan 12):
+The consolidated goal-era arc (plan 15, re-baselined for C+D):
 1. goal create -> start: the goal-run assignment reaches the daemon
    over the ask/allocate channel with the full plan-12 payload (kind/
    mode/creator/goal context/ephemeral branches for workers)
-2. pre-plan question block -> answered on the goal page -> the ONE
-   answer resolves the goal-run task's single task_blocks row
-   row, and the parked goal agent resumes with the verdict wording
+2. a commentary-only round and a trailing-comment round feed VERBATIM
+   (comment-policy-free inputs, plan 15 D1) before the pre-plan
+   question block -> answered on the goal page -> the ONE answer
+   resolves the goal-run task's single task_blocks row and the parked
+   goal agent resumes with the verdict wording
 3. plan submit -> reject-with-feedback -> revision -> approval (the
    agreed verdict words ride the resume)
 4. task fan-out through the real /api/saddle/task/create + enqueue
-   callbacks; the worker runs in its ephemeral worktree branch cut
-   from the goal clone, commits, completes — worktree auto-pruned,
-   branch kept
-5. the task-terminal event coalesces (~SADDLE_GOAL_RESUME_S) into the
-   goal agent's batched resume; it merges, deletes the task branch,
-   pushes the goal branch (push-before-review) and submits the
-   completion report
-6. completion review accept -> goal complete, run task done
+   callbacks; the worker's DEGENERATE first response is
+   validator-caught (plan 15 D2 — never executed, one receipt, the
+   session continues); the clean resend runs in its ephemeral
+   worktree branch cut from the goal clone, commits, completes —
+   worktree auto-pruned, branch kept
+5. the terminal transition (the goals row's last_terminal_watermark,
+   plan 15 C3 — goal_events is gone) coalesces (~SADDLE_GOAL_RESUME_S)
+   into the goal agent's batched resume; it merges, deletes the task
+   branch, pushes the goal branch (push-before-review) and submits
+   the completion report
+6. completion review accept — the live per-repo push flags gate it
+   (plan 15 C4): the accept REFUSES while origin is behind, passes
+   once the branch is re-pushed and the probe re-cached -> goal
+   complete, run task done
 7. the goal agent's terminal sessions (SSE replay), the audit trail
-   (the derived goal narrative, typed session_commands), the /status
-   goal_branches cache sampled by the manual saddle refresh
+   (the DERIVED goal narrative, typed session_commands, the
+   harness-authored system rows), the /status goal_branches cache
+   sampled by the manual saddle refresh
 8. the legacy ingest API surface (401s, typed updates, exactly-once
    poll handover) and the stm-markdown renderer
 
@@ -54,7 +63,8 @@ Environment:
   MOCK_PORT      local mock port       (default 9999)
   MOCK_CMD       run instead of the local mock (e.g. a podman exec line
                  that starts the mock inside the saddle container); the
-                 script skips spawning the mock when set
+                 script skips spawning the mock when set (the mock then
+                 needs its own --capture for the receipt assertions)
 
 Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
 """
@@ -79,6 +89,10 @@ DB_PATH = os.environ.get("DB_PATH", os.path.join(WORKSPACE, "..", "web", "clanke
 MOCK = os.environ.get("MOCK", os.path.join(os.path.dirname(REPO), "saddle", "test", "mock_llm.py"))
 MOCK_PORT = os.environ.get("MOCK_PORT", "9999")
 MOCK_CMD = os.environ.get("MOCK_CMD", "")
+# The mock's request capture (one JSON line per arrival carrying the
+# full message list) — the receipt forensics the D-side assertions read.
+CAPTURE_PATH = os.environ.get("CAPTURE_PATH",
+                              os.path.join(WORKSPACE, "..", "mock-capture.jsonl"))
 DEMO = os.path.join(WORKSPACE, "demo")
 HOSTNAME = socket.gethostname().split(".")[0]
 PLAN_APPROVED = "Plan approved. Proceed with implementation."
@@ -218,13 +232,31 @@ def saddle_api(method, path, token, payload=None):
 
 
 def db_query(sql, params=()):
-    db = sqlite3.connect(DB_PATH)
+    db = sqlite3.connect(DB_PATH, timeout=10)
     try:
         return db.execute(sql, params).fetchall()
     finally:
         db.close()
 
 
+def capture_user_turns():
+    """Every user turn the mock received, oldest first (the mock's
+    --capture file: one JSON line per request). The D-side receipts
+    (round outputs) ride these turns back to the model."""
+    turns = []
+    if os.path.exists(CAPTURE_PATH):
+        with open(CAPTURE_PATH) as handle:
+            for line in handle:
+                try:
+                    arrival = json.loads(line)
+                except ValueError:
+                    continue
+                for message in arrival.get("messages", []):
+                    if message.get("role") == "user":
+                        turns.append(message.get("content") or "")
+    return turns
+
+
 def prepare_workspace():
     """Ensures $WORKSPACE/demo exists with a pushable origin (the goal
     branch push needs a remote that accepts it). Reuses an existing
@@ -260,14 +292,27 @@ def mock_argv():
         "saddle suspend"
     )
     responses = [
-        # R1 (fresh goal agent): investigate, then block on a scope question
+        # R1 (fresh goal agent): commentary-ONLY (plan 15 D1) — fed
+        # verbatim, receipt [0 command lines] plus the factual
+        # no-commands line, conversation continues
+        "# Reading the brief before touching anything\n"
+        "# the demo repo is tiny; the greeting belongs in greeting.txt\n"
+        "# nothing worth running yet — probing comes next",
+        # R2: investigate with TRAILING comments after the commands
+        # (comments-after-commands feeds exactly like leading ones),
+        # then block on a scope question
         "# Investigating before planning\nls -la\ngit log --oneline | head -3\n"
+        "# two commits and one file — the blast radius is greeting.txt alone\n"
         + (heredoc % 'saddle block question "Demo only" "Whole workspace"') % (
             "Which scope should the greeting cover? The demo repo is small; confirm the blast radius."),
-        # R2 (plan revision after feedback): the rejected plan, re-submitted
+        # R3 (plan revision after feedback): the rejected plan, re-submitted
         (heredoc % "saddle plan") % (
             "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n2. Export it for reuse."),
-        # W1 (the worker task): verify the ephemeral branch, commit, done
+        # W1 (the worker's first round): a DEGENERATE repetition (plan
+        # 15 D2) — the validator catches it before feeding; nothing
+        # executes, one rejection receipt, the round is consumed
+        "# Starting the mandate\n" + "echo probing-the-workspace\n" * 12,
+        # W2 (the clean resend): verify the ephemeral branch, commit, done
         "# Working the mandate\ngit branch --show-current\n"
         "echo hello-from-the-goal > greeting.txt\ngit add -A\n"
         "git commit -m 'greeting from the goal'\n"
@@ -297,6 +342,9 @@ def mock_argv():
             "# Goal completion report\n\n- achieved: greeting.txt added via one worker task\n"
             "- verified: the goal branch is pushed to origin"),
         "--on-query", "The hostname is %s." % HOSTNAME,
+        # receipt forensics: every prompt the mock received, so the
+        # harness can assert the D-side receipts reached the model
+        "--capture", CAPTURE_PATH,
     ]
     return argv
 
@@ -365,6 +413,8 @@ def markdown_renderer_checks():
 
 def main():
     prepare_workspace()
+    if os.path.exists(CAPTURE_PATH):
+        os.remove(CAPTURE_PATH)
 
     if db_query("select count(*) from tasks")[0][0] or db_query("select count(*) from goals")[0][0]:
         raise SystemExit("the web database is not fresh (tasks/goals exist) — "
@@ -528,7 +578,7 @@ def main():
         print("== coalesced resume -> merge -> push -> completion review ==")
         check(wait_for(lambda: any(
             e["kind"] == "task-done" and int(e["task_id"]) == worker_task
-            for e in goal_state(goal1).get("events", [])) or False,
+            for e in goal_state(auth, goal1).get("events", [])) or False,
             timeout=30, what="the task-done narrative row") is not None,
               "the worker's terminal transition lands in the derived goal narrative")
         # the goal agent's /status reports the branch push state while it
@@ -540,10 +590,11 @@ def main():
         saddles_state = slot_public(auth, "/saddles", "saddles")
         card = [s for s in saddles_state["saddles"] if s["name"] == SADDLE_NAME][0]
         act(auth, card["refresh"], {})
-        check(wait_for(lambda: any("demo" in (b.get("branch") or "")
-                                   for b in goal_state(auth, goal1).get("push_branches", []))
-                       or False, timeout=60, what="cached push state") is not None,
-              "the refresh probe caches the branch push state on the goal")
+        check(wait_for(lambda: any(
+            "demo" in (b.get("branch") or "") and isinstance(b.get("pushed"), bool)
+            for b in goal_state(auth, goal1).get("push_branches", []))
+            or False, timeout=60, what="cached push state") is not None,
+              "the refresh probe caches the branch push state on the goal (live chips)")
         check(wait_goal_status(auth, goal1, "completion_review", timeout=120) is not None,
               "the resumed agent merges, pushes and enters completion review")
         state = goal_state(auth, goal1)
@@ -560,22 +611,60 @@ def main():
                                 capture_output=True).returncode == 0
         check(pushed, "the goal branch was pushed to origin (push-before-review)")
 
-        print("== completion accept ==")
+        print("== accept refuses while origin is behind (the live push gate) ==")
+        # Rewind origin's goal branch to the pre-merge base and refresh
+        # the daemon probe: the live per-repo flags flip to unpushed and
+        # the completion accept must REFUSE (plan 15 C4 / F9) until the
+        # branch is pushed again — enforcing push-after-each-merge at
+        # the human gate, read from probe state (never a snapshot).
         completion = state["completion_pending_block"]
+        subprocess.run(["git", "-C", goal_clone, "push", "-f", "origin",
+                        "main:refs/heads/goal/demo-run"], check=True, capture_output=True)
+        subprocess.run(["git", "-C", goal_clone, "fetch", "origin"],
+                       check=True, capture_output=True)
+        act(auth, card["refresh"], {})
+        check(wait_for(lambda: any(
+            "demo" in (b.get("branch") or "") and b.get("pushed") is False
+            for b in goal_state(auth, goal1).get("push_branches", [])) or False,
+            timeout=60, what="unpushed chip") is not None,
+              "the rewound origin flips the live push chip to unpushed")
+        refusal = act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
+        check(any(d.get("type") == "notify" and d.get("kind") == "error"
+                  and "Origin is behind" in (d.get("message") or "")
+                  for d in refusal),
+              "the accept refuses with a clear message while origin is behind")
+        check(goal_state(auth, goal1)["status"] == "completion_review",
+              "the refused accept leaves the goal in completion review")
+        check(goal_state(auth, goal1)["completion_pending_block"]
+              not in (None, False),
+              "the refused accept leaves the completion block unanswered")
+        subprocess.run(["git", "-C", goal_clone, "push", "origin", "goal/demo-run"],
+                       check=True, capture_output=True)
+        act(auth, card["refresh"], {})
+        check(wait_for(lambda: any(
+            "demo" in (b.get("branch") or "") and b.get("pushed") is True
+            for b in goal_state(auth, goal1).get("push_branches", [])) or False,
+            timeout=60, what="pushed chip restored") is not None,
+              "re-pushing the branch restores the live push chip (after push)")
+        completion = goal_state(auth, goal1)["completion_pending_block"]
+
+        print("== completion accept ==")
         check(completion and completion.get("body", "") == state.get("completion_report", ""),
               "the run task's completion block carries the report")
-        act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
+        accept = act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
+        check(not any(d.get("type") == "notify" and d.get("kind") == "error" for d in accept),
+              "the accept passes once the push flags read pushed")
         check(wait_goal_status(auth, goal1, "complete", timeout=30) is not None,
               "accepting with the agreed wording completes the goal")
         check(wait_task_status(auth, run_task, "done", timeout=30) is not None,
               "the goal-run task closes done with the goal")
 
         print("== audit trail (the derived narrative) ==")
-        narrative = goal_state(goal1)["events"]
+        narrative = goal_state(auth, goal1)["events"]
         kinds = [e["kind"] for e in narrative]
         for expected in ("started", "block-raised", "plan-submitted", "plan-rejected",
                          "plan-approved", "task-created", "task-done",
-                         "completion-submitted", "goal-approved"):
+                         "completion-submitted", "goal-approved", "system"):
             check(expected in kinds, "the derived narrative records %s" % expected)
         check(all(k != "task-enqueued" for k in kinds),
               "timestamp-less bookkeeping never returns (task-enqueued is gone)")
@@ -594,6 +683,70 @@ def main():
         check({"bash", "done"} <= worker_types,
               "the worker's terminal carries typed entries (%s)" % sorted(worker_types))
 
+        print("== the terminal watermark (C3) drives the resume ==")
+        watermark_row = db_query(
+            "select last_terminal_watermark from goals where id = ?", (goal1,))
+        worker_done_at = db_query(
+            "select completed_at from tasks where id = ?", (worker_task,))[0][0]
+        check(watermark_row[0][0] is not None and worker_done_at is not None
+              and watermark_row[0][0] >= worker_done_at,
+              "the goals row's terminal watermark advanced past the worker's transition"
+              " (the resume consumed it)")
+
+        print("== system rows on the ONE thread (C2) ==")
+        thread = goal_state(auth, goal1)["comments"]
+        system_rows = [c for c in thread if c.get("is_system")]
+        check(system_rows and all(c["author"] == "system" for c in system_rows)
+              and not any(c.get("from_agent") for c in system_rows)
+              and any("session context restored" in c["body"] for c in system_rows)
+              and any("parked until an owned task" in c["body"] for c in system_rows),
+              "the harness bookkeeping (restore, park) rides the thread as system rows,"
+              " never agent rows")
+        run_thread = task_state(auth, run_task)["comments"]
+        check(any(c.get("is_system") and "session context restored" in c["body"]
+                  for c in run_thread),
+              "the task page reads the same thread (the system rows included)")
+        check(not any(c.get("from_agent") and "[harness]" in c["body"]
+                      for c in thread),
+              "no harness-authored row renders as agent content")
+
+        print("== D-side through both binaries: commentary + degenerate ==")
+        run_bash = db_query(
+            "select sc.command_text from session_commands sc join sessions s"
+            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
+            (run_task,))
+        check(any("nothing worth running yet" in (row[0] or "")
+                  and "Reading the brief" in (row[0] or "") for row in run_bash),
+              "the narration-only response was fed verbatim (comments are style, not policy)")
+        check(any("git log --oneline" in (row[0] or "")
+                  and "blast radius is greeting.txt alone" in (row[0] or "")
+                  for row in run_bash),
+              "trailing comments after commands rode the same round (no rejection)")
+        rejected = db_query(
+            "select sc.command_text, sc.output_text from session_commands sc"
+            " join sessions s on s.id = sc.session_id"
+            " where sc.type = 'rejected' and s.task_id in (?, ?)",
+            (run_task, worker_task))
+        check(len(rejected) == 1 and rejected[0][0] == "degenerate response"
+              and "degenerate signature" in (rejected[0][1] or "")
+              and "echo probing-the-workspace" in (rejected[0][1] or "")
+              and "Resend the intended action only" in (rejected[0][1] or ""),
+              "exactly ONE rejected entry — the degenerate repetition, caught with its receipt")
+        worker_bash = db_query(
+            "select sc.command_text from session_commands sc join sessions s"
+            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
+            (worker_task,))
+        check(not any("probing-the-workspace" in (row[0] or "") for row in worker_bash),
+              "the degenerate script never executed (validator-caught before feeding)")
+        turns = capture_user_turns()
+        check(any(turn.startswith("[0 command lines executed")
+                  and "no commands ran — narrate in the same response as your next command" in turn
+                  for turn in turns),
+              "the narration round's receipt reached the model (no-commands line)")
+        check(any("degenerate signature" in turn and "Resend the intended action only" in turn
+                  for turn in turns),
+              "the degenerate rejection receipt reached the model as its next turn")
+
         print("== SSE ==")
         run_session = db_query(
             "select sc.session_id from session_commands sc join sessions s on s.id = sc.session_id"