| 1 | """Sync/divergence-recovery workflows: step contracts, skip behavior, and the
|
| 2 | error-actionability row fields.
|
| 3 |
|
| 4 | Oak commits are local checkpoints in the current launch contract. Remote sync
|
| 5 | workflows must publish explicitly with `oak push` or `oak commit --push`; the
|
| 6 | expected failure rows measure the remote command that fails, not the local
|
| 7 | snapshot.
|
| 8 | """
|
| 9 |
|
| 10 | from __future__ import annotations
|
| 11 |
|
| 12 | import sys
|
| 13 | import tempfile
|
| 14 | import unittest
|
| 15 | from pathlib import Path
|
| 16 | from unittest import mock
|
| 17 |
|
| 18 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 19 |
|
| 20 | import workflow_ab # noqa: E402
|
| 21 | from oakbench import integrity as oakbench_integrity # noqa: E402
|
| 22 | from oakbench import rows as oakbench_rows # noqa: E402
|
| 23 | from oakbench.subjects import Subject # noqa: E402
|
| 24 |
|
| 25 | GIT = Subject(name="git", kind="git", label="git", bin=Path("/usr/bin/git"))
|
| 26 | OAK = Subject(name="oak_installed", kind="oak", label="oak", bin=Path("oak"))
|
| 27 | CTX = {
|
| 28 | "branch": "bench-test-sync-r0-aa-bbbb",
|
| 29 | "remote_dir": "../remote-run-0.git",
|
| 30 | "clone_b": "../clone-b-run-0",
|
| 31 | "clone_a": "../clone-a-run-0",
|
| 32 | "repo_dir": "sync-push-divergence-r0",
|
| 33 | "measure_dir": "../clone-a-run-0",
|
| 34 | "oak_repo": "oak/bench-sync-tmp",
|
| 35 | }
|
| 36 | METADATA = {"bench_id": "test", "profile": "workflow-ab"}
|
| 37 |
|
| 38 |
|
| 39 | def operations(steps: list[workflow_ab.Step]) -> list[str]:
|
| 40 | return [step.operation for step in steps]
|
| 41 |
|
| 42 |
|
| 43 | class PushDivergenceStepsTest(unittest.TestCase):
|
| 44 | def test_git_recovery_preserves_work_no_redo_steps(self) -> None:
|
| 45 | steps = workflow_ab.sync_push_divergence_steps(GIT, "agent-default", CTX)
|
| 46 | ops = operations(steps)
|
| 47 | self.assertIn("error.push_diverged", ops)
|
| 48 | self.assertIn("error.pull_unconfigured", ops)
|
| 49 | self.assertIn("recovery.pull_rebase", ops)
|
| 50 | self.assertNotIn("edit.redo_change", ops)
|
| 51 |
|
| 52 | def test_git_expected_failure_codes_match_probed_contract(self) -> None:
|
| 53 | by_op = {s.operation: s for s in workflow_ab.sync_push_divergence_steps(GIT, "agent-default", CTX)}
|
| 54 | self.assertEqual(by_op["error.push_diverged"].expected_returncodes, (1,))
|
| 55 | self.assertEqual(by_op["error.pull_unconfigured"].expected_returncodes, (128,))
|
| 56 |
|
| 57 | def test_oak_recovery_includes_destructive_pull_and_redo(self) -> None:
|
| 58 | ops = operations(workflow_ab.sync_push_divergence_steps(OAK, "agent-default", CTX))
|
| 59 | # pull --force discards the local commit, so the workflow must redo
|
| 60 | # and re-snapshot the change; those steps ARE oak's recovery cost.
|
| 61 | self.assertIn("error.pull_diverged", ops)
|
| 62 | self.assertLess(ops.index("recovery.pull_force"), ops.index("edit.redo_change"))
|
| 63 | self.assertIn("vcs.snapshot.redo", ops)
|
| 64 |
|
| 65 | def test_oak_expected_failure_codes_match_probed_contract(self) -> None:
|
| 66 | by_op = {s.operation: s for s in workflow_ab.sync_push_divergence_steps(OAK, "agent-default", CTX)}
|
| 67 | self.assertEqual(by_op["vcs.snapshot"].expected_returncodes, (0,))
|
| 68 | self.assertEqual(by_op["vcs.snapshot"].command, ["oak", "commit", "--no-verify"])
|
| 69 | self.assertEqual(by_op["error.push_diverged"].expected_returncodes, (0, 1, 5))
|
| 70 | self.assertEqual(by_op["error.pull_diverged"].expected_returncodes, (5,))
|
| 71 |
|
| 72 | def test_oak_remote_setup_publishes_with_explicit_push(self) -> None:
|
| 73 | setup = {s.operation: " ".join(s.command) for s in workflow_ab.oak_clone_a_steps(OAK, CTX)}
|
| 74 |
|
| 75 | self.assertIn("commit --no-verify --push", setup["setup.remote.seed_main"])
|
| 76 | self.assertIn("commit --no-verify --push", setup["setup.remote.link"])
|
| 77 | self.assertLess(
|
| 78 | setup["setup.remote.link"].index("commit --no-verify --push"),
|
| 79 | setup["setup.remote.link"].index("branch rename"),
|
| 80 | )
|
| 81 |
|
| 82 | def test_oak_upstream_actor_publishes_with_explicit_push(self) -> None:
|
| 83 | setup = {s.operation: " ".join(s.command) for s in workflow_ab.sync_prepare_steps(OAK, "agent-default", CTX)}
|
| 84 |
|
| 85 | self.assertIn("commit --no-verify --push", setup["setup.upstream.move"])
|
| 86 |
|
| 87 | def test_error_steps_carry_recovery_hints(self) -> None:
|
| 88 | for subject in (GIT, OAK):
|
| 89 | for step in workflow_ab.sync_push_divergence_steps(subject, "agent-default", CTX):
|
| 90 | if step.operation.startswith("error."):
|
| 91 | self.assertIsNotNone(step.recovery_hint, step.operation)
|
| 92 |
|
| 93 | def test_both_subjects_converge_on_verify_payload(self) -> None:
|
| 94 | for subject in (GIT, OAK):
|
| 95 | ops = operations(workflow_ab.sync_push_divergence_steps(subject, "agent-default", CTX))
|
| 96 | self.assertEqual(ops[-1], "verify.payload")
|
| 97 |
|
| 98 | def test_pull_dirty_keeps_local_payload_check_after_pull(self) -> None:
|
| 99 | for subject in (GIT, OAK):
|
| 100 | steps = workflow_ab.sync_pull_dirty_steps(subject, "agent-default", CTX)
|
| 101 | by_op = {step.operation: step for step in steps}
|
| 102 | ops = operations(steps)
|
| 103 | self.assertEqual(by_op["sync.pull_dirty"].expected_returncodes, (0, 1, 2, 4, 5))
|
| 104 | self.assertLess(ops.index("sync.pull_dirty"), ops.index("verify.local_payload"))
|
| 105 | self.assertIn("dirty local edit", " ".join(by_op["verify.local_payload"].command))
|
| 106 |
|
| 107 | def test_test_steps_disable_python_bytecode_writes(self) -> None:
|
| 108 | step = workflow_ab.Step("test.green", "test", workflow_ab.shell("python3 -m unittest"))
|
| 109 | env = workflow_ab.step_env(step)
|
| 110 | self.assertEqual(env["PYTHONDONTWRITEBYTECODE"], "1")
|
| 111 |
|
| 112 | def test_push_divergence_snapshot_requires_only_local_payload(self) -> None:
|
| 113 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 114 | requirements = workflow_ab.workflow_step_payload_requirements(workflow, "vcs.snapshot.commit")
|
| 115 | self.assertEqual(requirements, [("work.txt", "local change")])
|
| 116 |
|
| 117 | with tempfile.TemporaryDirectory() as tmp:
|
| 118 | repo = Path(tmp)
|
| 119 | (repo / "work.txt").write_text("base line\nlocal change\n")
|
| 120 | (repo / "notes.txt").write_text("shared notes\n")
|
| 121 | fields = oakbench_integrity.content_payload_fields(repo, requirements)
|
| 122 |
|
| 123 | self.assertTrue(fields["content_payload_check_passed"])
|
| 124 | self.assertEqual(fields["content_payload_missing_paths"], [])
|
| 125 |
|
| 126 | def test_final_push_divergence_integrity_still_requires_upstream_payload(self) -> None:
|
| 127 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 128 | with tempfile.TemporaryDirectory() as tmp:
|
| 129 | repo = Path(tmp)
|
| 130 | (repo / "work.txt").write_text("base line\nlocal change\n")
|
| 131 | (repo / "notes.txt").write_text("shared notes\n")
|
| 132 | fields = workflow_ab.workflow_integrity_fields(GIT, workflow, repo, [])
|
| 133 |
|
| 134 | self.assertFalse(fields["workflow_integrity_passed"])
|
| 135 | self.assertIn("notes.txt", fields["workflow_integrity_failure_reason"])
|
| 136 |
|
| 137 |
|
| 138 | class SyncContextTest(unittest.TestCase):
|
| 139 | def test_oak_without_env_yields_skip_reason_naming_env_var(self) -> None:
|
| 140 | import os
|
| 141 |
|
| 142 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 143 | scrubbed = {
|
| 144 | k: v for k, v in os.environ.items() if k not in ("OAK_BENCH_SYNC_REPO", "OAK_BENCH_REMOTE")
|
| 145 | }
|
| 146 | with mock.patch.dict("os.environ", scrubbed, clear=True):
|
| 147 | ctx, reason, fields = workflow_ab.sync_context(OAK, workflow, 0, METADATA)
|
| 148 | self.assertIsNone(ctx)
|
| 149 | self.assertIn("OAK_BENCH_SYNC_REPO", reason or "")
|
| 150 | self.assertEqual(fields, {})
|
| 151 |
|
| 152 | def test_oak_with_env_resolves_network_transport(self) -> None:
|
| 153 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 154 | with mock.patch.dict("os.environ", {"OAK_BENCH_SYNC_REPO": "oak/bench-sync-tmp"}), mock.patch.object(
|
| 155 | workflow_ab, "oak_commit_push_skip_reason", return_value=None
|
| 156 | ):
|
| 157 | ctx, reason, fields = workflow_ab.sync_context(OAK, workflow, 3, METADATA)
|
| 158 | self.assertIsNone(reason)
|
| 159 | self.assertEqual(ctx["oak_repo"], "oak/bench-sync-tmp")
|
| 160 | self.assertEqual(fields["remote_transport"], "network")
|
| 161 | self.assertIn("-r3-", ctx["branch"])
|
| 162 |
|
| 163 | def test_oak_with_old_cli_skips_explicit_publish_workflows(self) -> None:
|
| 164 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 165 | with mock.patch.dict("os.environ", {"OAK_BENCH_SYNC_REPO": "oak/bench-sync-tmp"}), mock.patch.object(
|
| 166 | workflow_ab,
|
| 167 | "oak_commit_push_skip_reason",
|
| 168 | return_value="oak_commit_push_not_supported:requires_current_oak",
|
| 169 | ):
|
| 170 | ctx, reason, fields = workflow_ab.sync_context(OAK, workflow, 3, METADATA)
|
| 171 | self.assertIsNone(ctx)
|
| 172 | self.assertEqual(reason, "oak_commit_push_not_supported:requires_current_oak")
|
| 173 | self.assertEqual(fields["remote_transport"], "network")
|
| 174 |
|
| 175 | def test_git_resolves_local_file_transport(self) -> None:
|
| 176 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 177 | ctx, reason, fields = workflow_ab.sync_context(GIT, workflow, 0, METADATA)
|
| 178 | self.assertIsNone(reason)
|
| 179 | self.assertEqual(fields["remote_transport"], "local_file")
|
| 180 | self.assertEqual(ctx["remote_dir"], "../remote-run-0.git")
|
| 181 |
|
| 182 |
|
| 183 | class WorkflowSkipRowTest(unittest.TestCase):
|
| 184 | def test_skip_row_satisfies_workflow_lane_contract(self) -> None:
|
| 185 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 186 | row = workflow_ab.workflow_skip_row(OAK, workflow, 0, METADATA, "oak_remote_not_configured: set OAK_BENCH_SYNC_REPO")
|
| 187 | errors = oakbench_rows.validate_rows([row], "workflow")
|
| 188 | self.assertEqual(errors, [])
|
| 189 | self.assertEqual(row["returncode"], 77)
|
| 190 | self.assertTrue(row["skipped"])
|
| 191 |
|
| 192 |
|
| 193 | class WorkflowSummaryTransportTest(unittest.TestCase):
|
| 194 | def test_time_deltas_are_not_reported_across_unlike_transports(self) -> None:
|
| 195 | workflow = workflow_ab.WORKFLOWS["sync_pull_upstream"]
|
| 196 | rows = [
|
| 197 | {
|
| 198 | **METADATA,
|
| 199 | "benchmark_track": "agent-default",
|
| 200 | "subject": "git",
|
| 201 | "workflow": workflow.name,
|
| 202 | "operation": "workflow.total",
|
| 203 | "returncode": 0,
|
| 204 | "elapsed_ms": 50.0,
|
| 205 | "estimated_tokens_total": 100,
|
| 206 | "raw_output_bytes": 1000,
|
| 207 | "tool_call_count": 3,
|
| 208 | "remote_repo": "../remote.git",
|
| 209 | "remote_transport": "local_file",
|
| 210 | "remote_server": "local",
|
| 211 | },
|
| 212 | {
|
| 213 | **METADATA,
|
| 214 | "benchmark_track": "agent-default",
|
| 215 | "subject": "oak_installed",
|
| 216 | "workflow": workflow.name,
|
| 217 | "operation": "workflow.total",
|
| 218 | "returncode": 0,
|
| 219 | "elapsed_ms": 500.0,
|
| 220 | "estimated_tokens_total": 80,
|
| 221 | "raw_output_bytes": 800,
|
| 222 | "tool_call_count": 2,
|
| 223 | "remote_repo": "oak/bench-sync-tmp",
|
| 224 | "remote_transport": "network",
|
| 225 | "remote_server": "oak.space",
|
| 226 | },
|
| 227 | ]
|
| 228 |
|
| 229 | summary = workflow_ab.summary_text(rows, [GIT, OAK], [workflow])
|
| 230 |
|
| 231 | self.assertIn("Time deltas are not comparable across different remote/workspace transports", summary)
|
| 232 | oak_line = next(line for line in summary.splitlines() if "| `sync_pull_upstream` | `oak_installed` |" in line)
|
| 233 | self.assertIn("| n/c transport |", oak_line)
|
| 234 | self.assertIn("| +20.0% |", oak_line)
|
| 235 |
|
| 236 |
|
| 237 | class ErrorActionabilityRowTest(unittest.TestCase):
|
| 238 | def test_row_records_whether_error_mentions_recovery_command(self) -> None:
|
| 239 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 240 | with tempfile.TemporaryDirectory() as tmp:
|
| 241 | mentioned = workflow_ab.run_step(
|
| 242 | GIT,
|
| 243 | workflow,
|
| 244 | workflow_ab.Step(
|
| 245 | "error.fake", "vcs",
|
| 246 | workflow_ab.shell("echo 'hint: use git pull first' >&2; exit 1"),
|
| 247 | (1,),
|
| 248 | recovery_hint="git pull",
|
| 249 | ),
|
| 250 | Path(tmp), 0, "workflow", METADATA, 20_000,
|
| 251 | )
|
| 252 | missing = workflow_ab.run_step(
|
| 253 | GIT,
|
| 254 | workflow,
|
| 255 | workflow_ab.Step(
|
| 256 | "error.fake", "vcs",
|
| 257 | workflow_ab.shell("echo 'something failed' >&2; exit 1"),
|
| 258 | (1,),
|
| 259 | recovery_hint="git pull",
|
| 260 | ),
|
| 261 | Path(tmp), 0, "workflow", METADATA, 20_000,
|
| 262 | )
|
| 263 | self.assertTrue(mentioned["error_mentions_recovery_command"])
|
| 264 | self.assertEqual(mentioned["returncode"], 0) # expected failure is on-script
|
| 265 | self.assertFalse(missing["error_mentions_recovery_command"])
|
| 266 |
|
| 267 | def test_expected_failure_that_returns_zero_is_harness_failure(self) -> None:
|
| 268 | workflow = workflow_ab.WORKFLOWS["sync_push_divergence"]
|
| 269 | with tempfile.TemporaryDirectory() as tmp:
|
| 270 | row = workflow_ab.run_step(
|
| 271 | GIT,
|
| 272 | workflow,
|
| 273 | workflow_ab.Step("error.fails_to_fail", "vcs", workflow_ab.shell("exit 0"), (1,)),
|
| 274 | Path(tmp), 0, "workflow", METADATA, 20_000,
|
| 275 | )
|
| 276 | self.assertEqual(row["process_returncode"], 0)
|
| 277 | self.assertEqual(row["returncode"], 1)
|
| 278 |
|
| 279 |
|
| 280 | if __name__ == "__main__":
|
| 281 | unittest.main()
|