feat(1e): HC3 additive generic + op/assertion split (orchestrator owns the op)

- orchestrator: per mutating tier, run optional pre-op seed hook (ops.py pre_<op>) → perform the op
  ONCE (harness-owned) → run generic assertion (unless opted out) AND overlay assertion, both against
  the shared post-op deployment. Op results passed op→assertion via run-scoped CCCI_OP_STATE_FILE.
- opt-out: CCCI_SKIP_GENERIC / CCCI_SKIP_GENERIC_<OP> / recipe_meta.SKIP_GENERIC (declarative).
- generic.py: split do_* into op primitives (perform_upgrade/backup/restore) + assertions
  (assert_upgraded/backup_artifact/restore_healthy) reading op_state(); deployed_identity now returns
  {version,image,chaos} (chaos label ready for HC1).
- generic test_<op>.py + all 6 recipe overlays migrated to assertion-only; pre-op seeding moved to
  per-recipe ops.py (pre_upgrade/pre_backup/pre_restore). install overlays unchanged (no op).
- deploy-count stays 1 (op primitives never call deploy_app). lint PASS; 8 unit tests PASS on cc-ci.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

runner/harness/generic.py

@@ -118,22 +118,49 @@ def assert_serving(domain: str, meta: dict) -> None:
     assert "commoninternet.net" in detail.lower(), f"{domain}: served cert unexpected — {detail}"
 
 
-def do_upgrade(domain: str, target: str | None, meta: dict) -> None:
-    """UPGRADE op (in place on the shared deployment): abra app upgrade -> target, then assert it
-    reconverges + still serves AND that the deployment actually MOVED (version label and/or image
-    changed). The move assertion guards against a vacuous no-op upgrade silently passing — the exact
-    F1d-2 failure where a mis-pinned base deployed LATEST so 'upgrade to latest' changed nothing."""
-    before = lifecycle.deployed_identity(domain)
-    lifecycle.upgrade_app(domain, version=target)
+# ---- Op/assertion split (Phase 1e HC3) -------------------------------------------------------
+# The orchestrator performs each mutating op ONCE (the harness owns the op), records what an
+# assertion needs (pre-upgrade identity, backup snapshot_id) into a run-scoped JSON state file at
+# $CCCI_OP_STATE_FILE, then runs the generic assertion file (unless opted out) AND the overlay
+# assertion file against the shared post-op state. The assertion functions below read that state via
+# `op_state()`. They NEVER perform the op — that keeps the op single + lets generic+overlay coexist.
+
+import json as _json  # noqa: E402
+
+
+def op_state() -> dict:
+    """The run-scoped op state the orchestrator wrote between op and assertions (or {} if unset).
+    Carries e.g. {"upgrade": {"before": {...}}, "backup": {"snapshot_id": "..."}}."""
+    path = os.environ.get("CCCI_OP_STATE_FILE")
+    if not path or not os.path.exists(path):
+        return {}
+    try:
+        with open(path) as f:
+            return _json.load(f)
+    except (OSError, ValueError):
+        return {}
+
+
+def assert_upgraded(domain: str, meta: dict) -> None:
+    """Generic UPGRADE assertion (post-op): the orchestrator already performed the upgrade once.
+    Assert it reconverged + still serves AND that the deployment actually MOVED — guarding against a
+    vacuous no-op upgrade silently passing (F1d-2). HC1: prev→PR-head may NOT bump the version label,
+    so a MOVE is ANY of: version-label change, image change, or a chaos label now present (a chaos
+    deploy stamps the PR-head commit — THE proof the code under test was deployed)."""
+    before = op_state().get("upgrade", {}).get("before") or {}
     assert_serving(domain, meta)
     after = lifecycle.deployed_identity(domain)
-    moved = (before[0] and after[0] and before[0] != after[0]) or (
-        before[1] and after[1] and before[1] != after[1]
+    moved = (
+        (before.get("version") and after.get("version") and before["version"] != after["version"])
+        or (before.get("image") and after.get("image") and before["image"] != after["image"])
+        or (after.get("chaos") and after.get("chaos") != before.get("chaos"))
     )
     assert moved, (
         f"{domain}: upgrade did not move the deployment "
-        f"(version {before[0]}->{after[0]}, image {before[1]}->{after[1]}) — "
-        "not a real previous->target upgrade (DG2 must be non-vacuous)"
+        f"(version {before.get('version')}->{after.get('version')}, "
+        f"image {before.get('image')}->{after.get('image')}, "
+        f"chaos {before.get('chaos')}->{after.get('chaos')}) — "
+        "not a real upgrade to the code under test (HC1/DG2 must be non-vacuous)"
     )
 
 
@@ -148,10 +175,10 @@ def parse_snapshot_id(backup_output: str) -> str | None:
     return m.group(1) if m else None
 
 
-def do_backup(domain: str) -> str:
-    """BACKUP op: create a backup, then assert a snapshot artifact was produced (returns its id)."""
-    out = lifecycle.backup_app(domain)
-    snap_id = parse_snapshot_id(out)
+def assert_backup_artifact(domain: str) -> str:
+    """Generic BACKUP assertion (post-op): the orchestrator already ran the backup once. Assert a
+    snapshot artifact was produced (its id recorded in op state). Returns the id."""
+    snap_id = op_state().get("backup", {}).get("snapshot_id")
     assert snap_id, (
         f"{domain}: backup produced no snapshot artifact "
         "(no snapshot_id in `abra app backup create` output)"
@@ -159,8 +186,29 @@ def do_backup(domain: str) -> str:
     return snap_id
 
 
-def do_restore(domain: str, meta: dict) -> None:
-    """RESTORE op: restore the latest snapshot, then assert the app is healthy + serving again
-    (assert_serving polls, so the post-restore reconverge settles)."""
-    lifecycle.restore_app(domain)
+def assert_restore_healthy(domain: str, meta: dict) -> None:
+    """Generic RESTORE assertion (post-op): the orchestrator already restored. Assert the app is
+    healthy + serving again (assert_serving polls, so the post-restore reconverge settles)."""
     assert_serving(domain, meta)
+
+
+# ---- Op primitives (orchestrator-only; perform the op once, never assert) --------------------
+
+
+def perform_upgrade(domain: str, target: str | None) -> dict[str, str | None]:
+    """Perform the UPGRADE op once (in place). E1 baseline: `abra app upgrade` -> target. (HC1/E2
+    redefines this as a chaos redeploy of the PR-head checkout.) Returns the pre-upgrade identity so
+    the orchestrator can record it for `assert_upgraded`'s move check."""
+    before = lifecycle.deployed_identity(domain)
+    lifecycle.upgrade_app(domain, version=target)
+    return before
+
+
+def perform_backup(domain: str) -> str | None:
+    """Perform the BACKUP op once. Returns the produced snapshot_id (or None) for the assertion."""
+    return parse_snapshot_id(lifecycle.backup_app(domain))
+
+
+def perform_restore(domain: str) -> None:
+    """Perform the RESTORE op once (restore the latest snapshot)."""
+    lifecycle.restore_app(domain)