diff --git a/docs/RELEASE_CONSOLE.md b/docs/RELEASE_CONSOLE.md index 6ddef4d..6c09bd7 100644 --- a/docs/RELEASE_CONSOLE.md +++ b/docs/RELEASE_CONSOLE.md @@ -63,38 +63,47 @@ these digests detect accidental or manual corruption, not an attacker who can replace the private record and recompute its checksums. Changing a target, channel, or gate input requires a new run. -Run records live below `runtime/release-runs/`. They survive console restarts, -but remain local operator state rather than a signed release artifact or the -system audit log. The directory is restricted to mode `0700` and records to -`0600`. Writes use a cross-process lock, a same-directory temporary file, -`fsync`, atomic replacement, and directory `fsync`. Symbolic-link storage paths, -overly broad record modes, malformed schemas, unknown fields, invalid state -combinations, oversized files, and digest mismatches fail closed. A bad record -is neither rewritten nor automatically quarantined. +Run records survive console restarts, but remain local operator state rather +than a signed release artifact or the system audit log. The default location is +`$XDG_STATE_HOME/govoplan/release-console/workspace-/release-runs/`, or +`~/.local/state/...` when `XDG_STATE_HOME` is unset or relative. It is outside +the source checkout so filesystems without enforceable POSIX modes cannot +silently weaken the journal. Newly created state directories use mode `0700` +and records use `0600`. Writes use a cross-process lock, a same-directory +temporary file, `fsync`, atomic replacement, and directory/parent `fsync`. +Symbolic-link paths, untrusted owners or writable ancestry, overly broad record +modes, malformed schemas, unknown fields, invalid state combinations, +oversized files, and digest mismatches fail closed. A bad record is neither +rewritten nor automatically quarantined. -The default store belongs to the selected workspace: the console uses that -workspace's meta checkout when it is available, or a stable workspace-digest -namespace below this checkout's runtime directory otherwise. The same digest is -part of the immutable input snapshot. An alternate workspace therefore cannot -silently list or resume another workspace's runs. `run_state_root` remains an -explicit embedding/test override. +The full resolved-workspace SHA-256 is part of both the private storage +namespace and immutable input snapshot. Every list, read, and state transition +checks it. An alternate workspace therefore cannot list or resume another +workspace's runs even when an embedding/test `run_state_root` override is +shared accidentally. Each frozen plan step has an explicit `pending`, `running`, `succeeded`, `failed`, or `interrupted` state. Plan order remains a prerequisite: a later step is unavailable until earlier steps have succeeded. Exact attempt and -resume/retry request identifiers are fingerprinted so repeated state commands -are idempotent. The record keeps at most 256 server-generated state events and -128 request fingerprints. Events contain only an enum event type, timestamp, -step identifier, and bounded result code—never commands, process output, -confirmation text, credentials, bearer tokens, or signing material. +resume/retry/reconciliation request identifiers are fingerprinted so delayed +repetitions remain idempotent for the run's lifetime. These fingerprints are +never evicted: the store fails closed before accepting more than 2,048 commands +or 2,048 attempts and asks the operator to create a fresh run. The display +record keeps at most 256 server-generated state events. Events contain only an +enum event type, timestamp, step identifier, and bounded result code—never +commands, process output, confirmation text, credentials, bearer tokens, or +signing material. An explicit resume after a process restart converts every persisted `running` step to `interrupted`; it never guesses whether an external effect happened. An interrupted read-only step can be prepared for retry. An interrupted -mutating step remains unavailable until the operator has reconciled local and -remote state and explicitly records that reconciliation when preparing the -retry. A known failed attempt can likewise be prepared for retry. The UI keeps -the retry control visible and explains why it is disabled when retry is unsafe. +mutating step remains unavailable until the operator independently reconciles +local and remote state, selects `effect_absent`, `effect_succeeded`, or +`unresolved`, and types `RECONCILE`. `effect_absent` prepares a safe new +attempt, `effect_succeeded` advances the run without repeating the external +effect, and `unresolved` keeps the run blocked. Each outcome emits a bounded, +code-only state event. A known failed attempt can likewise be prepared for +retry. The UI keeps unavailable controls visible and disabled. The run API is covered by the same local console token middleware as every other `/api/` route: @@ -105,7 +114,12 @@ other `/api/` route: - `GET /api/release-runs/{run_id}` reads and verifies one exact record. - `POST /api/release-runs/{run_id}/resume` records explicit recovery. - `POST /api/release-runs/{run_id}/steps/{step_id}/retry` prepares a failed or - safely reconciled interrupted step for another attempt. + read-only interrupted step for another attempt. +- `POST /api/release-runs/{run_id}/steps/{step_id}/reconcile` records a + confirmed observed outcome for an interrupted mutating step. + +Run-storage errors are confined to the Durable Run State card; dashboard and +release-preview collection continue and show the bounded storage remediation. This foundation tracks state only. It deliberately does not run a plan step or infer success from a button click. Existing preview and executor controls stay diff --git a/tests/test_release_run_state.py b/tests/test_release_run_state.py index 9b91c98..f0c5cea 100644 --- a/tests/test_release_run_state.py +++ b/tests/test_release_run_state.py @@ -22,7 +22,9 @@ from govoplan_release.release_run import ( # noqa: E402 MAX_EVENTS, ReleaseRunConflict, ReleaseRunCorrupt, + ReleaseRunNotFound, ReleaseRunStore, + _record_digest, ) from server.app import ( # noqa: E402 create_app, @@ -31,11 +33,17 @@ from server.app import ( # noqa: E402 ) +WORKSPACE_FINGERPRINT = "a" * 64 + + class ReleaseRunStoreTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) / "release-runs" - self.store = ReleaseRunStore(self.root) + self.store = ReleaseRunStore( + self.root, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ) def tearDown(self) -> None: self.temporary.cleanup() @@ -45,11 +53,14 @@ class ReleaseRunStoreTests(unittest.TestCase): input_snapshot=run_input(), plan_snapshot=release_plan() ) - reopened = ReleaseRunStore(self.root).get(run["run_id"]) + reopened = ReleaseRunStore( + self.root, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ).get(run["run_id"]) record_path = self.root / f"{run['run_id']}.json" self.assertEqual("govoplan.release-run", reopened["schema"]) - self.assertEqual(1, reopened["schema_version"]) + self.assertEqual(2, reopened["schema_version"]) self.assertEqual( {"govoplan-core": "1.2.3", "govoplan-files": "1.2.4"}, reopened["immutable"]["input"]["repo_versions"], @@ -63,6 +74,12 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertNotIn("processed_requests", reopened["state"]) self.assertNotIn("attempt_fingerprint", reopened["state"]["steps"][0]) + def test_zero_step_plan_is_rejected(self) -> None: + plan = release_plan() + plan["dry_run_steps"] = [] + with self.assertRaisesRegex(ReleaseRunConflict, "step collection"): + self.store.create(input_snapshot=run_input(), plan_snapshot=plan) + def test_attempt_transitions_are_idempotent_and_strictly_ordered(self) -> None: run_id = self.store.create( input_snapshot=run_input(), plan_snapshot=release_plan() @@ -104,13 +121,119 @@ class ReleaseRunStoreTests(unittest.TestCase): ) self.assertTrue(finished["state"]["steps"][1]["available"]) + def test_failed_attempt_and_delayed_start_replays_are_idempotent(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + attempt_id = "failed-attempt-replay-0001" + self.store.start_step(run_id, "core:preflight", attempt_id=attempt_id) + failed = self.store.finish_step( + run_id, + "core:preflight", + attempt_id=attempt_id, + succeeded=False, + result_code="preflight_failed", + ) + duplicate = self.store.start_step( + run_id, "core:preflight", attempt_id=attempt_id + ) + self.assertEqual("failed", duplicate["state"]["steps"][0]["state"]) + self.assertEqual( + len(failed["state"]["events"]), len(duplicate["state"]["events"]) + ) + + retried = self.store.retry_step( + run_id, + "core:preflight", + request_id="failed-retry-request-0001", + ) + delayed = self.store.start_step( + run_id, "core:preflight", attempt_id=attempt_id + ) + self.assertEqual("pending", delayed["state"]["steps"][0]["state"]) + self.assertEqual( + len(retried["state"]["events"]), len(delayed["state"]["events"]) + ) + + def test_delayed_resume_request_is_idempotent_for_the_run_lifetime(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + original_request = "resume-original-request-0001" + self.store.resume(run_id, request_id=original_request) + for index in range(130): + self.store.resume( + run_id, request_id=f"resume-intervening-request-{index:04d}" + ) + running = self.store.start_step( + run_id, + "core:preflight", + attempt_id="resume-delayed-attempt-0001", + ) + duplicate = self.store.resume(run_id, request_id=original_request) + self.assertEqual("running", duplicate["state"]["steps"][0]["state"]) + self.assertEqual( + len(running["state"]["events"]), len(duplicate["state"]["events"]) + ) + + def test_command_and_attempt_capacity_fail_closed_without_eviction(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + with patch("govoplan_release.release_run.MAX_COMMANDS", 2): + self.store.resume(run_id, request_id="capacity-resume-request-0001") + self.store.resume(run_id, request_id="capacity-resume-request-0002") + with self.assertRaisesRegex(ReleaseRunConflict, "safety limit"): + self.store.resume(run_id, request_id="capacity-resume-request-0003") + duplicate = self.store.resume( + run_id, request_id="capacity-resume-request-0001" + ) + self.assertEqual("planned", duplicate["state"]["status"]) + + attempt_run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + with patch("govoplan_release.release_run.MAX_COMMANDS", 1): + self.store.start_step( + attempt_run_id, + "core:preflight", + attempt_id="capacity-attempt-0001", + ) + self.store.finish_step( + attempt_run_id, + "core:preflight", + attempt_id="capacity-attempt-0001", + succeeded=False, + result_code="preflight_failed", + ) + self.store.retry_step( + attempt_run_id, + "core:preflight", + request_id="capacity-retry-request-0001", + ) + with self.assertRaisesRegex(ReleaseRunConflict, "attempt history"): + self.store.start_step( + attempt_run_id, + "core:preflight", + attempt_id="capacity-attempt-0002", + ) + delayed_attempt = self.store.start_step( + attempt_run_id, + "core:preflight", + attempt_id="capacity-attempt-0001", + ) + self.assertEqual("pending", delayed_attempt["state"]["steps"][0]["state"]) + def test_reopen_and_resume_preserve_ambiguous_mutating_effect(self) -> None: run_id = self.store.create( input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="attempt-tag-0001") - reopened_store = ReleaseRunStore(self.root) + reopened_store = ReleaseRunStore( + self.root, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ) before_resume = reopened_store.get(run_id) self.assertEqual("running", before_resume["state"]["steps"][0]["state"]) @@ -118,28 +241,38 @@ class ReleaseRunStoreTests(unittest.TestCase): duplicate = reopened_store.resume(run_id, request_id="resume-request-0001") self.assertEqual("interrupted", resumed["state"]["steps"][0]["state"]) self.assertEqual("reconcile_step", resumed["recommended_next"]["id"]) - self.assertFalse(resumed["recommended_next"]["available"]) + self.assertTrue(resumed["recommended_next"]["available"]) self.assertEqual( len(resumed["state"]["events"]), len(duplicate["state"]["events"]) ) - with self.assertRaisesRegex(ReleaseRunConflict, "must be reconciled"): + with self.assertRaisesRegex(ReleaseRunConflict, "observed effect"): reopened_store.retry_step( run_id, "core:tag", request_id="retry-request-0001", ) - retried = reopened_store.retry_step( + with self.assertRaisesRegex(ReleaseRunConflict, "Type RECONCILE"): + reopened_store.reconcile_step( + run_id, + "core:tag", + request_id="reconcile-request-0001", + outcome="effect_absent", + confirmation="", + ) + retried = reopened_store.reconcile_step( run_id, "core:tag", - request_id="retry-request-0002", - reconciled=True, + request_id="reconcile-request-0002", + outcome="effect_absent", + confirmation="RECONCILE", ) - duplicate_retry = reopened_store.retry_step( + duplicate_retry = reopened_store.reconcile_step( run_id, "core:tag", - request_id="retry-request-0002", - reconciled=True, + request_id="reconcile-request-0002", + outcome="effect_absent", + confirmation="RECONCILE", ) self.assertEqual("pending", retried["state"]["steps"][0]["state"]) self.assertEqual( @@ -147,6 +280,47 @@ class ReleaseRunStoreTests(unittest.TestCase): len(duplicate_retry["state"]["events"]), ) + def test_reconciliation_can_record_unresolved_or_verified_success(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=mutating_first_plan() + )["run_id"] + self.store.start_step(run_id, "core:tag", attempt_id="reconcile-attempt-0001") + self.store.resume(run_id, request_id="reconcile-resume-request-0001") + + unresolved = self.store.reconcile_step( + run_id, + "core:tag", + request_id="reconcile-outcome-request-0001", + outcome="unresolved", + confirmation="RECONCILE", + ) + self.assertEqual("interrupted", unresolved["state"]["steps"][0]["state"]) + self.assertEqual( + "unresolved", unresolved["state"]["events"][-1]["result_code"] + ) + with self.assertRaisesRegex(ReleaseRunConflict, "different run-state action"): + self.store.reconcile_step( + run_id, + "core:tag", + request_id="reconcile-outcome-request-0001", + outcome="effect_succeeded", + confirmation="RECONCILE", + ) + + succeeded = self.store.reconcile_step( + run_id, + "core:tag", + request_id="reconcile-outcome-request-0002", + outcome="effect_succeeded", + confirmation="RECONCILE", + ) + self.assertEqual("succeeded", succeeded["state"]["steps"][0]["state"]) + self.assertEqual( + "reconciled_effect_succeeded", + succeeded["state"]["steps"][0]["result_code"], + ) + self.assertEqual("completed", succeeded["state"]["status"]) + def test_tampered_record_fails_closed_and_is_not_rewritten(self) -> None: run_id = self.store.create( input_snapshot=run_input(), plan_snapshot=release_plan() @@ -178,7 +352,10 @@ class ReleaseRunStoreTests(unittest.TestCase): linked = Path(self.temporary.name) / "linked" linked.symlink_to(target, target_is_directory=True) with self.assertRaisesRegex(ReleaseRunCorrupt, "symbolic-link"): - ReleaseRunStore(linked).list() + ReleaseRunStore( + linked, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ).list() def test_state_specific_fields_and_boolean_counts_fail_closed(self) -> None: run_id = self.store.create( @@ -218,6 +395,110 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertEqual(tampered, path.read_bytes()) + def test_impossible_step_chronology_fails_even_with_recomputed_checksum(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + self.store.start_step( + run_id, "core:preflight", attempt_id="chronology-attempt-0001" + ) + self.store.finish_step( + run_id, + "core:preflight", + attempt_id="chronology-attempt-0001", + succeeded=True, + result_code="preflight_valid", + ) + path = self.root / f"{run_id}.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["state"]["steps"][0]["started_at"] = "2026-07-22T20:00:00Z" + payload["state"]["steps"][0]["finished_at"] = "2026-07-22T19:00:00Z" + payload["record_digest"] = _record_digest(payload) + path.write_text(json.dumps(payload), encoding="utf-8") + path.chmod(0o600) + + with self.assertRaises(ReleaseRunCorrupt): + self.store.get(run_id) + + def test_record_and_retained_event_timestamps_are_strictly_ordered(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + self.store.resume(run_id, request_id="timestamp-resume-request-0001") + path = self.root / f"{run_id}.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["created_at"] = "2026-07-22T00:00:00Z" + payload["updated_at"] = "2026-07-22T00:00:03Z" + payload["state"]["events"][0]["at"] = "2026-07-22T00:00:02Z" + payload["state"]["events"][1]["at"] = "2026-07-22T00:00:01Z" + payload["record_digest"] = _record_digest(payload) + path.write_text(json.dumps(payload), encoding="utf-8") + path.chmod(0o600) + + with self.assertRaises(ReleaseRunCorrupt): + self.store.get(run_id) + + def test_attempt_count_must_match_lifetime_attempt_history(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + self.store.start_step( + run_id, "core:preflight", attempt_id="counted-attempt-0001" + ) + path = self.root / f"{run_id}.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["state"]["steps"][0]["attempt_count"] = 2 + payload["record_digest"] = _record_digest(payload) + path.write_text(json.dumps(payload), encoding="utf-8") + path.chmod(0o600) + + with self.assertRaises(ReleaseRunCorrupt): + self.store.get(run_id) + + def test_workspace_fingerprint_is_enforced_for_every_store_operation(self) -> None: + run_id = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + )["run_id"] + foreign = ReleaseRunStore( + self.root, + expected_workspace_fingerprint="b" * 64, + ) + + self.assertEqual([], foreign.list()) + with self.assertRaises(ReleaseRunNotFound): + foreign.get(run_id) + with self.assertRaises(ReleaseRunNotFound): + foreign.resume(run_id, request_id="foreign-resume-request-0001") + with self.assertRaisesRegex(ReleaseRunConflict, "different workspace"): + foreign.create(input_snapshot=run_input(), plan_snapshot=release_plan()) + + def test_list_limit_is_applied_after_foreign_workspace_records_are_skipped( + self, + ) -> None: + with patch( + "govoplan_release.release_run._timestamp", + return_value="2026-07-22T00:00:00Z", + ): + local_run = self.store.create( + input_snapshot=run_input(), plan_snapshot=release_plan() + ) + foreign_store = ReleaseRunStore( + self.root, + expected_workspace_fingerprint="b" * 64, + ) + foreign_input = run_input() + foreign_input["workspace_fingerprint"] = "b" * 64 + with patch( + "govoplan_release.release_run._timestamp", + return_value="2026-07-22T00:00:01Z", + ): + foreign_store.create( + input_snapshot=foreign_input, plan_snapshot=release_plan() + ) + + listed = self.store.list(limit=1) + self.assertEqual([local_run["run_id"]], [item["run_id"] for item in listed]) + def test_concurrent_claims_have_one_winner_and_no_lost_update(self) -> None: run_id = self.store.create( input_snapshot=run_input(), plan_snapshot=release_plan() @@ -225,7 +506,10 @@ class ReleaseRunStoreTests(unittest.TestCase): barrier = threading.Barrier(2) def claim(attempt_id: str) -> str: - contender = ReleaseRunStore(self.root) + contender = ReleaseRunStore( + self.root, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ) barrier.wait(timeout=5) try: contender.start_step(run_id, "core:preflight", attempt_id=attempt_id) @@ -250,7 +534,7 @@ class ReleaseRunStoreTests(unittest.TestCase): ), ) - def test_event_and_request_history_are_bounded_and_code_only(self) -> None: + def test_event_history_is_bounded_and_command_history_is_not_evicted(self) -> None: run_id = self.store.create( input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] @@ -273,7 +557,8 @@ class ReleaseRunStoreTests(unittest.TestCase): view = self.store.get(run_id) raw = json.loads((self.root / f"{run_id}.json").read_text(encoding="utf-8")) self.assertEqual(MAX_EVENTS, len(view["state"]["events"])) - self.assertLessEqual(len(raw["state"]["processed_requests"]), 128) + self.assertEqual(90, len(raw["state"]["processed_requests"])) + self.assertEqual(90, len(raw["state"]["processed_attempts"])) self.assertNotIn("bounded-retry-0089", json.dumps(raw)) for event in view["state"]["events"]: self.assertLessEqual( @@ -369,7 +654,10 @@ class ReleaseRunApiTests(unittest.TestCase): def test_api_rejects_tampered_record_without_rewriting_it(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" - store = ReleaseRunStore(root) + store = ReleaseRunStore( + root, + expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, + ) run_id = store.create( input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] @@ -388,26 +676,119 @@ class ReleaseRunApiTests(unittest.TestCase): self.assertEqual(409, response.status_code) self.assertEqual(malformed, path.read_bytes()) + def test_api_maps_unavailable_run_storage_without_server_exception(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + parent = Path(temp_dir) + target = parent / "target" + target.mkdir() + linked_root = parent / "linked-runs" + linked_root.symlink_to(target, target_is_directory=True) + app = create_app( + workspace_root=parent, + run_state_root=linked_root, + token="token", + ) + with TestClient(app) as client: + response = client.get( + "/api/release-runs", + headers={"X-Release-Console-Token": "token"}, + ) + self.assertEqual(409, response.status_code) + self.assertIn("symbolic-link", response.json()["detail"]) + + def test_api_requires_confirmation_and_records_reconciliation_outcome(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) / "runs" + app = create_app( + workspace_root=Path(temp_dir), run_state_root=root, token="token" + ) + headers = {"X-Release-Console-Token": "token"} + with ( + patch("server.app.build_dashboard", return_value=object()), + patch( + "server.app.build_selective_release_plan", + return_value=mutating_first_plan(), + ), + TestClient(app) as client, + ): + created = client.post( + "/api/release-runs", headers=headers, json=run_input() + ).json() + run_id = created["run_id"] + app.state.release_runs.start_step( + run_id, "core:tag", attempt_id="api-reconcile-attempt-0001" + ) + client.post( + f"/api/release-runs/{run_id}/resume", + headers=headers, + json={"request_id": "api-reconcile-resume-0001"}, + ) + missing_confirmation = client.post( + f"/api/release-runs/{run_id}/steps/core:tag/reconcile", + headers=headers, + json={ + "request_id": "api-reconcile-request-0001", + "outcome": "effect_succeeded", + }, + ) + reconciled = client.post( + f"/api/release-runs/{run_id}/steps/core:tag/reconcile", + headers=headers, + json={ + "request_id": "api-reconcile-request-0002", + "outcome": "effect_succeeded", + "confirm": "RECONCILE", + }, + ) + + self.assertEqual(409, missing_confirmation.status_code) + self.assertEqual(200, reconciled.status_code) + self.assertEqual("completed", reconciled.json()["state"]["status"]) + self.assertEqual( + "effect_succeeded", + reconciled.json()["state"]["events"][-1]["result_code"], + ) + def test_default_storage_and_immutable_input_are_workspace_bound(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: parent = Path(temp_dir) + state_home = parent / "operator-state" first = parent / "first" second = parent / "second" first.mkdir() second.mkdir() - first_root = default_release_run_root(first) - second_root = default_release_run_root(second) + with patch.dict( + os.environ, {"XDG_STATE_HOME": str(state_home)}, clear=False + ): + first_root = default_release_run_root(first) + second_root = default_release_run_root(second) + default_app = create_app(workspace_root=first, token="token") + with TestClient(default_app) as client: + default_list = client.get( + "/api/release-runs", + headers={"X-Release-Console-Token": "token"}, + ) + self.assertEqual(200, default_list.status_code) + self.assertEqual(0o700, first_root.stat().st_mode & 0o777) self.assertNotEqual(first_root, second_root) - self.assertIn(release_workspace_fingerprint(first)[:16], first_root.name) + self.assertEqual("release-runs", first_root.name) + self.assertEqual(state_home, first_root.parents[3]) + self.assertEqual( + f"workspace-{release_workspace_fingerprint(first)}", + first_root.parent.name, + ) + self.assertFalse(first_root.is_relative_to(first)) local_workspace = parent / "local" local_meta = local_workspace / "govoplan" local_meta.mkdir(parents=True) (local_meta / "repositories.json").write_text("{}", encoding="utf-8") - self.assertEqual( - local_meta / "runtime" / "release-runs", - default_release_run_root(local_workspace), - ) + with patch.dict( + os.environ, {"XDG_STATE_HOME": str(state_home)}, clear=False + ): + self.assertTrue( + default_release_run_root(local_workspace).is_relative_to(state_home) + ) explicit_root = parent / "explicit" app = create_app( @@ -435,6 +816,41 @@ class ReleaseRunApiTests(unittest.TestCase): ) self.assertEqual(explicit_root.resolve(), app.state.release_runs.root) + def test_shared_explicit_root_does_not_cross_workspace_boundary(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + parent = Path(temp_dir) + root = parent / "shared-runs" + first = parent / "first" + second = parent / "second" + first.mkdir() + second.mkdir() + first_app = create_app( + workspace_root=first, run_state_root=root, token="token" + ) + headers = {"X-Release-Console-Token": "token"} + with ( + patch("server.app.build_dashboard", return_value=object()), + patch( + "server.app.build_selective_release_plan", + return_value=release_plan(), + ), + TestClient(first_app) as client, + ): + created = client.post( + "/api/release-runs", headers=headers, json=run_input() + ).json() + + second_app = create_app( + workspace_root=second, run_state_root=root, token="token" + ) + with TestClient(second_app) as client: + listed = client.get("/api/release-runs", headers=headers) + loaded = client.get( + f"/api/release-runs/{created['run_id']}", headers=headers + ) + self.assertEqual([], listed.json()["runs"]) + self.assertEqual(404, loaded.status_code) + def test_ui_and_runbook_state_tracking_boundary_are_explicit(self) -> None: webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") runbook = (META_ROOT / "docs" / "RELEASE_CONSOLE.md").read_text( @@ -446,9 +862,14 @@ class ReleaseRunApiTests(unittest.TestCase): self.assertIn("Create Run from Selection", webui) self.assertIn("Prepare Retry", webui) self.assertIn("step.retry_available", webui) + self.assertIn("Record Reconciliation", webui) + self.assertIn("effect_succeeded", webui) + self.assertIn("renderReleaseRunStoreUnavailable", webui) + self.assertIn('const runsRequest = api("/api/release-runs")', webui) self.assertIn("This foundation tracks state only.", runbook) - self.assertIn("mutating step remains unavailable", runbook) - self.assertIn("same-directory temporary file", runbook) + self.assertIn("effect_absent", runbook) + self.assertIn("$XDG_STATE_HOME", runbook) + self.assertIn("same-directory", runbook) def run_input() -> dict[str, object]: @@ -462,7 +883,7 @@ def run_input() -> dict[str, object]: "remote_tags": False, "public_catalog": False, "include_migrations": True, - "workspace_fingerprint": "a" * 64, + "workspace_fingerprint": WORKSPACE_FINGERPRINT, } diff --git a/tools/release/govoplan_release/release_run.py b/tools/release/govoplan_release/release_run.py index 434d7d4..792aa77 100644 --- a/tools/release/govoplan_release/release_run.py +++ b/tools/release/govoplan_release/release_run.py @@ -21,16 +21,16 @@ import threading from typing import Any import uuid -from filelock import FileLock +from filelock import FileLock, Timeout as FileLockTimeout SCHEMA = "govoplan.release-run" -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 MAX_RECORD_BYTES = 2 * 1024 * 1024 MAX_PLAN_STEPS = 512 MAX_REPOSITORY_VERSIONS = 128 MAX_EVENTS = 256 -MAX_REQUESTS = 128 +MAX_COMMANDS = 2_048 MAX_LIST_LIMIT = 100 RUN_ID_PATTERN = re.compile(r"^rr-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}$") STEP_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,159}$") @@ -51,8 +51,12 @@ EVENT_TYPES = { "step_succeeded", "step_failed", "step_interrupted", + "step_reconciled", "step_retry_requested", } +ATTEMPT_OUTCOMES = {"running", "succeeded", "failed", "interrupted"} +RECONCILIATION_OUTCOMES = {"effect_absent", "effect_succeeded", "unresolved"} +RECONCILIATION_CONFIRMATION = "RECONCILE" class ReleaseRunError(RuntimeError): @@ -63,6 +67,10 @@ class ReleaseRunNotFound(ReleaseRunError): """Raised when a run does not exist.""" +class ReleaseRunWorkspaceMismatch(ReleaseRunNotFound): + """Raised when a valid record belongs to another workspace.""" + + class ReleaseRunConflict(ReleaseRunError): """Raised when a state transition is unsafe or no longer applicable.""" @@ -74,10 +82,15 @@ class ReleaseRunCorrupt(ReleaseRunError): class ReleaseRunStore: """Atomic, bounded local JSON store for release-run state.""" - def __init__(self, root: Path) -> None: + def __init__(self, root: Path, *, expected_workspace_fingerprint: str) -> None: # Keep the configured path spelling so a symlink root (or symlinked # parent) remains detectable instead of being erased by resolve(). self.root = Path(os.path.abspath(os.fspath(root.expanduser()))) + if not WORKSPACE_FINGERPRINT_PATTERN.fullmatch( + expected_workspace_fingerprint + ): + raise ReleaseRunConflict("Release workspace fingerprint is invalid.") + self.expected_workspace_fingerprint = expected_workspace_fingerprint self._thread_lock = threading.RLock() self._lock_path = self.root / ".release-runs.lock" @@ -88,6 +101,13 @@ class ReleaseRunStore: plan_snapshot: dict[str, Any], ) -> dict[str, Any]: immutable_input = _validated_input(input_snapshot) + if ( + immutable_input["workspace_fingerprint"] + != self.expected_workspace_fingerprint + ): + raise ReleaseRunConflict( + "Release run input belongs to a different workspace." + ) immutable_plan = _validated_plan(plan_snapshot) now = _timestamp() run_id = ( @@ -123,9 +143,11 @@ class ReleaseRunStore: "events": [], "next_event_sequence": 1, "processed_requests": [], + "processed_attempts": [], }, } _append_event(record, event_type="run_created") + record["updated_at"] = record["state"]["events"][-1]["at"] _refresh_status(record) record["record_digest"] = _record_digest(record) with self._locked(): @@ -151,11 +173,13 @@ class ReleaseRunStore: ), key=lambda item: item.name, reverse=True, - )[:limit] + ) summaries: list[dict[str, Any]] = [] for path in paths: try: record = self._read(path.stem) + except ReleaseRunWorkspaceMismatch: + continue except (ReleaseRunCorrupt, ReleaseRunNotFound): summaries.append( { @@ -164,6 +188,8 @@ class ReleaseRunStore: "error": "Run record is unavailable or failed integrity validation.", } ) + if len(summaries) >= limit: + break continue view = _view(record) immutable_input = view["immutable"]["input"] @@ -178,17 +204,25 @@ class ReleaseRunStore: "recommended_next": view["recommended_next"], } ) + if len(summaries) >= limit: + break return summaries def resume(self, run_id: str, *, request_id: str) -> dict[str, Any]: request_fingerprint = _request_fingerprint(request_id) with self._locked(): record = self._read(run_id) - if _request_seen(record, request_fingerprint, "resume", None): + if _request_seen(record, request_fingerprint, "resume", None, None): return _view(record) + _ensure_command_capacity(record) for step in record["state"]["steps"]: if step["state"] != "running": continue + attempt = _attempt_for_fingerprint( + record, step["attempt_fingerprint"], step_id=step["id"] + ) + attempt["outcome"] = "interrupted" + attempt["result_code"] = "process_interrupted" step["state"] = "interrupted" step["finished_at"] = _timestamp() step["result_code"] = "process_interrupted" @@ -198,7 +232,7 @@ class ReleaseRunStore: step_id=step["id"], result_code="process_interrupted", ) - _remember_request(record, request_fingerprint, "resume", None) + _remember_request(record, request_fingerprint, "resume", None, None) _append_event(record, event_type="run_resumed") self._persist_update(record) return _view(record) @@ -209,27 +243,28 @@ class ReleaseRunStore: step_id: str, *, request_id: str, - reconciled: bool = False, ) -> dict[str, Any]: _validate_step_id(step_id) request_fingerprint = _request_fingerprint(request_id) with self._locked(): record = self._read(run_id) - if _request_seen(record, request_fingerprint, "retry", step_id): + if _request_seen(record, request_fingerprint, "retry", step_id, None): return _view(record) step, plan_step = _step_pair(record, step_id) if step["state"] not in {"failed", "interrupted"}: raise ReleaseRunConflict( "Only a failed or interrupted release step can be prepared for retry." ) - if ( - step["state"] == "interrupted" - and bool(plan_step.get("mutating")) - and not reconciled - ): + if step["state"] == "interrupted" and bool(plan_step.get("mutating")): raise ReleaseRunConflict( - "An interrupted mutating step must be reconciled before retry." + "Record the observed effect of an interrupted mutating step before retry." ) + _ensure_command_capacity(record) + retry_reason = ( + "known_failure" + if step["state"] == "failed" + else "read_only_interruption" + ) step.update( { "state": "pending", @@ -239,12 +274,75 @@ class ReleaseRunStore: "result_code": None, } ) - _remember_request(record, request_fingerprint, "retry", step_id) + _remember_request(record, request_fingerprint, "retry", step_id, None) _append_event( record, event_type="step_retry_requested", step_id=step_id, - result_code="reconciled" if reconciled else "known_failure", + result_code=retry_reason, + ) + self._persist_update(record) + return _view(record) + + def reconcile_step( + self, + run_id: str, + step_id: str, + *, + request_id: str, + outcome: str, + confirmation: str, + ) -> dict[str, Any]: + """Record an operator-observed outcome for an interrupted mutation.""" + + _validate_step_id(step_id) + if outcome not in RECONCILIATION_OUTCOMES: + raise ReleaseRunConflict("Release reconciliation outcome is invalid.") + if confirmation != RECONCILIATION_CONFIRMATION: + raise ReleaseRunConflict( + f"Type {RECONCILIATION_CONFIRMATION} to record reconciliation." + ) + request_fingerprint = _request_fingerprint(request_id) + with self._locked(): + record = self._read(run_id) + if _request_seen( + record, request_fingerprint, "reconcile", step_id, outcome + ): + return _view(record) + step, plan_step = _step_pair(record, step_id) + if step["state"] != "interrupted" or not bool( + plan_step.get("mutating") + ): + raise ReleaseRunConflict( + "Only an interrupted mutating step can be reconciled." + ) + _ensure_command_capacity(record) + if outcome == "effect_absent": + step.update( + { + "state": "pending", + "attempt_fingerprint": None, + "started_at": None, + "finished_at": None, + "result_code": None, + } + ) + elif outcome == "effect_succeeded": + step.update( + { + "state": "succeeded", + "finished_at": _timestamp(), + "result_code": "reconciled_effect_succeeded", + } + ) + _remember_request( + record, request_fingerprint, "reconcile", step_id, outcome + ) + _append_event( + record, + event_type="step_reconciled", + step_id=step_id, + result_code=outcome, ) self._persist_update(record) return _view(record) @@ -265,15 +363,18 @@ class ReleaseRunStore: attempt_fingerprint = _request_fingerprint(attempt_id) with self._locked(): record = self._read(run_id) - step, _plan_step = _step_pair(record, step_id) - if step["attempt_fingerprint"] == attempt_fingerprint and step["state"] in { - "running", - "succeeded", - }: + existing_attempt = _find_attempt(record, attempt_fingerprint) + if existing_attempt is not None: + if existing_attempt["step_id"] != step_id: + raise ReleaseRunConflict( + "The attempt identifier was already used for another release step." + ) return _view(record) + step, _plan_step = _step_pair(record, step_id) available, reason = _step_available(record, step_id) if not available: raise ReleaseRunConflict(reason) + _ensure_attempt_capacity(record) step.update( { "state": "running", @@ -284,6 +385,14 @@ class ReleaseRunStore: "result_code": None, } ) + record["state"]["processed_attempts"].append( + { + "fingerprint": attempt_fingerprint, + "step_id": step_id, + "outcome": "running", + "result_code": None, + } + ) _append_event(record, event_type="step_started", step_id=step_id) self._persist_update(record) return _view(record) @@ -306,12 +415,20 @@ class ReleaseRunStore: with self._locked(): record = self._read(run_id) step, _plan_step = _step_pair(record, step_id) - if ( - step["attempt_fingerprint"] == attempt_fingerprint - and step["state"] == target_state - and step["result_code"] == result_code - ): - return _view(record) + attempt = _find_attempt(record, attempt_fingerprint) + if attempt is None or attempt["step_id"] != step_id: + raise ReleaseRunConflict( + "The release attempt identifier was not claimed for this step." + ) + if attempt["outcome"] != "running": + if ( + attempt["outcome"] == target_state + and attempt["result_code"] == result_code + ): + return _view(record) + raise ReleaseRunConflict( + "The release attempt already has a different recorded outcome." + ) if ( step["state"] != "running" or step["attempt_fingerprint"] != attempt_fingerprint @@ -326,6 +443,8 @@ class ReleaseRunStore: "result_code": result_code, } ) + attempt["outcome"] = target_state + attempt["result_code"] = result_code _append_event( record, event_type="step_succeeded" if succeeded else "step_failed", @@ -336,6 +455,7 @@ class ReleaseRunStore: return _view(record) def _persist_update(self, record: dict[str, Any]) -> None: + self._assert_workspace(record) record["updated_at"] = _timestamp() _refresh_status(record) record["record_digest"] = _record_digest(record) @@ -366,6 +486,10 @@ class ReleaseRunStore: metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode): raise ReleaseRunCorrupt("Release run record is not a regular file.") + if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): + raise ReleaseRunCorrupt( + "Release run record is not owned by the console operator." + ) if metadata.st_mode & 0o077: raise ReleaseRunCorrupt( "Release run record permissions are broader than 0600." @@ -384,8 +508,16 @@ class ReleaseRunStore: if descriptor >= 0: os.close(descriptor) _validate_record(payload, expected_run_id=run_id) + self._assert_workspace(payload) return payload + def _assert_workspace(self, record: dict[str, Any]) -> None: + if ( + record["immutable"]["input"]["workspace_fingerprint"] + != self.expected_workspace_fingerprint + ): + raise ReleaseRunWorkspaceMismatch("Release run was not found.") + def _path(self, run_id: str) -> Path: if not RUN_ID_PATTERN.fullmatch(run_id): raise ReleaseRunNotFound("Release run was not found.") @@ -393,10 +525,15 @@ class ReleaseRunStore: def _ensure_root(self) -> None: _reject_symlink_components(self.root) - self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + _create_private_directory_tree(self.root) _reject_symlink_components(self.root) if self.root.is_symlink() or not self.root.is_dir(): raise ReleaseRunCorrupt("Release run storage root is not a directory.") + root_metadata = self.root.stat() + if hasattr(os, "geteuid") and root_metadata.st_uid != os.geteuid(): + raise ReleaseRunCorrupt( + "Release run storage is not owned by the console operator." + ) try: self.root.chmod(0o700) except OSError as exc: @@ -407,6 +544,7 @@ class ReleaseRunStore: raise ReleaseRunCorrupt( "Release run storage permissions are broader than 0700." ) + _fsync_directory(self.root) def _atomic_write(self, path: Path, record: dict[str, Any]) -> None: encoded = _canonical_json(record) + b"\n" @@ -426,20 +564,19 @@ class ReleaseRunStore: os.fsync(handle.fileno()) os.replace(temporary_path, path) temporary_path = None - try: - path.chmod(0o600) - except OSError as exc: + written_metadata = path.lstat() + if ( + not stat.S_ISREG(written_metadata.st_mode) + or written_metadata.st_mode & 0o077 + or ( + hasattr(os, "geteuid") + and written_metadata.st_uid != os.geteuid() + ) + ): raise ReleaseRunCorrupt( "Release run record permissions could not be secured." - ) from exc - directory_flags = os.O_RDONLY - if hasattr(os, "O_DIRECTORY"): - directory_flags |= os.O_DIRECTORY - directory_descriptor = os.open(self.root, directory_flags) - try: - os.fsync(directory_descriptor) - finally: - os.close(directory_descriptor) + ) + _fsync_directory(self.root) finally: if descriptor >= 0: os.close(descriptor) @@ -463,6 +600,11 @@ class _CombinedLock: self.thread_lock.acquire() try: self.file_lock.acquire(timeout=10) + except FileLockTimeout as exc: + self.thread_lock.release() + raise ReleaseRunConflict( + "Release run storage is busy; retry the operation." + ) from exc except BaseException: self.thread_lock.release() raise @@ -563,7 +705,11 @@ def _validated_plan(value: dict[str, Any]) -> dict[str, Any]: if recommendation is not None and not isinstance(recommendation, dict): raise ReleaseRunConflict("Release plan recommendation is invalid.") steps = copied.get("dry_run_steps") - if not isinstance(steps, list) or len(steps) > MAX_PLAN_STEPS: + if ( + not isinstance(steps, list) + or not steps + or len(steps) > MAX_PLAN_STEPS + ): raise ReleaseRunConflict("Release plan step collection is invalid.") seen: set[str] = set() for step in steps: @@ -630,6 +776,10 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: for key in ("created_at", "updated_at"): if not _is_timestamp(value.get(key)): raise ValueError + created_at = value["created_at"] + updated_at = value["updated_at"] + if updated_at < created_at: + raise ValueError immutable = value.get("immutable") if not isinstance(immutable, dict) or set(immutable) != { "input", @@ -664,6 +814,7 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: "events", "next_event_sequence", "processed_requests", + "processed_attempts", }: raise ValueError steps = state.get("steps") @@ -703,10 +854,22 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: if result_code is not None: _safe_code(result_code) _validate_step_state_fields(step) + for timestamp in (step.get("started_at"), step.get("finished_at")): + if timestamp is not None and not ( + created_at <= timestamp <= updated_at + ): + raise ValueError + first_incomplete_seen = False + for step in steps: + if first_incomplete_seen and step["state"] != "pending": + raise ValueError + if step["state"] != "succeeded": + first_incomplete_seen = True events = state.get("events") if not isinstance(events, list) or len(events) > MAX_EVENTS: raise ValueError previous_sequence = 0 + previous_event_at: str | None = None for event in events: if not isinstance(event, dict) or event.get("type") not in EVENT_TYPES: raise ValueError @@ -723,6 +886,11 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: previous_sequence = sequence if not _is_timestamp(event.get("at")): raise ValueError + if not (created_at <= event["at"] <= updated_at): + raise ValueError + if previous_event_at is not None and event["at"] < previous_event_at: + raise ValueError + previous_event_at = event["at"] if event.get("step_id") is not None: _validate_step_id(event["step_id"]) if event["step_id"] not in plan_step_ids: @@ -734,14 +902,16 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: if type(next_sequence) is not int or next_sequence <= previous_sequence: raise ValueError requests = state.get("processed_requests") - if not isinstance(requests, list) or len(requests) > MAX_REQUESTS: + if not isinstance(requests, list) or len(requests) > MAX_COMMANDS: raise ValueError for request in requests: if ( not isinstance(request, dict) - or set(request) != {"fingerprint", "operation", "step_id"} + or set(request) + != {"fingerprint", "operation", "step_id", "parameter"} or not _is_sha256(request.get("fingerprint")) - or request.get("operation") not in {"resume", "retry"} + or request.get("operation") + not in {"resume", "retry", "reconcile"} ): raise ValueError if request.get("step_id") is not None: @@ -750,8 +920,68 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: raise ValueError if (request["operation"] == "resume") != (request["step_id"] is None): raise ValueError + parameter = request["parameter"] + if request["operation"] == "reconcile": + if parameter not in RECONCILIATION_OUTCOMES: + raise ValueError + elif parameter is not None: + raise ValueError if len({request["fingerprint"] for request in requests}) != len(requests): raise ValueError + attempts = state.get("processed_attempts") + if not isinstance(attempts, list) or len(attempts) > MAX_COMMANDS: + raise ValueError + for attempt in attempts: + if ( + not isinstance(attempt, dict) + or set(attempt) + != {"fingerprint", "step_id", "outcome", "result_code"} + or not _is_sha256(attempt.get("fingerprint")) + or attempt.get("outcome") not in ATTEMPT_OUTCOMES + ): + raise ValueError + _validate_step_id(attempt.get("step_id")) + if attempt["step_id"] not in plan_step_ids: + raise ValueError + attempt_result = attempt.get("result_code") + if attempt["outcome"] == "running": + if attempt_result is not None: + raise ValueError + else: + if attempt_result is None: + raise ValueError + _safe_code(attempt_result) + if ( + attempt["outcome"] == "interrupted" + and attempt_result != "process_interrupted" + ): + raise ValueError + if len({attempt["fingerprint"] for attempt in attempts}) != len(attempts): + raise ValueError + attempts_by_fingerprint = { + attempt["fingerprint"]: attempt for attempt in attempts + } + attempts_per_step = { + step_id: sum( + attempt["step_id"] == step_id for attempt in attempts + ) + for step_id in plan_step_ids + } + for step in steps: + if step["attempt_count"] != attempts_per_step[step["id"]]: + raise ValueError + fingerprint = step["attempt_fingerprint"] + if fingerprint is None: + continue + attempt = attempts_by_fingerprint.get(fingerprint) + if attempt is None or attempt["step_id"] != step["id"]: + raise ValueError + expected_outcome = step["state"] + if step["result_code"] == "reconciled_effect_succeeded": + if step["state"] != "succeeded" or attempt["outcome"] != "interrupted": + raise ValueError + elif attempt["outcome"] != expected_outcome: + raise ValueError if state.get("status") != _status(value): raise ValueError if value.get("record_digest") != _record_digest(value): @@ -791,11 +1021,14 @@ def _view(record: dict[str, Any]) -> dict[str, Any]: "order": index + 1, "available": available, "retry_available": retry_available, + "reconciliation_required": step["state"] == "interrupted" + and bool(plan_step.get("mutating")), "disabled_reason": disabled_reason, } ) view["state"]["steps"] = enriched_steps view["state"].pop("processed_requests", None) + view["state"].pop("processed_attempts", None) view["recommended_next"] = _recommended_next(record) return view @@ -876,7 +1109,7 @@ def _recommended_next(record: dict[str, Any]) -> dict[str, Any]: step, plan_step, "Reconcile the local and remote effect before explicitly preparing a retry.", - False, + True, ) return _step_recommendation( "retry_step", @@ -960,23 +1193,44 @@ def _append_event( record["state"]["events"] = record["state"]["events"][-MAX_EVENTS:] +def _ensure_command_capacity(record: dict[str, Any]) -> None: + if len(record["state"]["processed_requests"]) >= MAX_COMMANDS: + raise ReleaseRunConflict( + "Release run command history reached its safety limit; create a new run." + ) + + +def _ensure_attempt_capacity(record: dict[str, Any]) -> None: + if len(record["state"]["processed_attempts"]) >= MAX_COMMANDS: + raise ReleaseRunConflict( + "Release run attempt history reached its safety limit; create a new run." + ) + + def _remember_request( - record: dict[str, Any], fingerprint: str, operation: str, step_id: str | None + record: dict[str, Any], + fingerprint: str, + operation: str, + step_id: str | None, + parameter: str | None, ) -> None: + _ensure_command_capacity(record) record["state"]["processed_requests"].append( { "fingerprint": fingerprint, "operation": operation, "step_id": step_id, + "parameter": parameter, } ) - record["state"]["processed_requests"] = record["state"]["processed_requests"][ - -MAX_REQUESTS: - ] def _request_seen( - record: dict[str, Any], fingerprint: str, operation: str, step_id: str | None + record: dict[str, Any], + fingerprint: str, + operation: str, + step_id: str | None, + parameter: str | None, ) -> bool: matches = [ request @@ -986,7 +1240,9 @@ def _request_seen( if not matches: return False if any( - request["operation"] == operation and request["step_id"] == step_id + request["operation"] == operation + and request["step_id"] == step_id + and request["parameter"] == parameter for request in matches ): return True @@ -995,6 +1251,34 @@ def _request_seen( ) +def _find_attempt( + record: dict[str, Any], fingerprint: str +) -> dict[str, Any] | None: + return next( + ( + attempt + for attempt in record["state"]["processed_attempts"] + if attempt["fingerprint"] == fingerprint + ), + None, + ) + + +def _attempt_for_fingerprint( + record: dict[str, Any], fingerprint: str | None, *, step_id: str +) -> dict[str, Any]: + if fingerprint is None: + raise ReleaseRunCorrupt( + "Running release step has no persisted attempt identifier." + ) + attempt = _find_attempt(record, fingerprint) + if attempt is None or attempt["step_id"] != step_id: + raise ReleaseRunCorrupt( + "Running release step has no matching persisted attempt." + ) + return attempt + + def _refresh_status(record: dict[str, Any]) -> None: record["state"]["status"] = _status(record) @@ -1088,6 +1372,8 @@ def _validate_step_state_fields(step: dict[str, Any]) -> None: raise ValueError if finished_at is not None and not _is_timestamp(finished_at): raise ValueError + if started_at is not None and finished_at is not None and finished_at < started_at: + raise ValueError if state == "pending": if any( value is not None @@ -1123,6 +1409,110 @@ def _validate_event_fields(event: dict[str, Any]) -> None: return if not has_result: raise ValueError + if event_type == "step_reconciled" and event["result_code"] not in ( + RECONCILIATION_OUTCOMES + ): + raise ValueError + if ( + event_type == "step_interrupted" + and event["result_code"] != "process_interrupted" + ): + raise ValueError + + +def _create_private_directory_tree(path: Path) -> None: + missing: list[Path] = [] + current = path + while not current.exists(): + missing.append(current) + if current == current.parent: + break + current = current.parent + _validate_authority_components(current) + for directory in reversed(missing): + try: + os.mkdir(directory, mode=0o700) + except FileExistsError: + pass + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run storage directory could not be created safely." + ) from exc + _reject_symlink_components(directory) + try: + metadata = directory.lstat() + if not stat.S_ISDIR(metadata.st_mode): + raise ReleaseRunCorrupt( + "Release run storage path is not a directory." + ) + if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): + raise ReleaseRunCorrupt( + "Release run storage is not owned by the console operator." + ) + directory.chmod(0o700) + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run storage permissions could not be secured." + ) from exc + if directory.stat().st_mode & 0o077: + raise ReleaseRunCorrupt( + "Release run storage permissions are broader than 0700." + ) + _fsync_directory(directory) + _fsync_directory(directory.parent) + _validate_authority_components(path.parent) + + +def _validate_authority_components(path: Path) -> None: + home = Path(os.path.abspath(os.fspath(Path.home()))) + if path == home or path.is_relative_to(home): + current = home + parts = path.relative_to(home).parts + candidates = [current] + else: + current = Path(path.anchor) + parts = path.parts[1:] + candidates = [] + for part in parts: + current /= part + candidates.append(current) + for current in candidates: + try: + metadata = current.lstat() + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run storage ancestry could not be validated." + ) from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise ReleaseRunCorrupt( + "Release run storage ancestry is not a safe directory path." + ) + if hasattr(os, "geteuid") and metadata.st_uid not in {0, os.geteuid()}: + raise ReleaseRunCorrupt( + "Release run storage ancestry has an untrusted owner." + ) + writable_by_others = bool(metadata.st_mode & 0o022) + sticky = bool(metadata.st_mode & stat.S_ISVTX) + if writable_by_others and not sticky: + raise ReleaseRunCorrupt( + "Release run storage ancestry is writable by another principal." + ) + + +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + try: + descriptor = os.open(path, flags) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run storage metadata could not be persisted." + ) from exc def _reject_symlink_components(path: Path) -> None: diff --git a/tools/release/server/app.py b/tools/release/server/app.py index 6685f47..4f70da1 100644 --- a/tools/release/server/app.py +++ b/tools/release/server/app.py @@ -3,7 +3,9 @@ from __future__ import annotations import hashlib +import os from pathlib import Path +from typing import Literal from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse @@ -108,7 +110,12 @@ class ReleaseRunCreateRequest(BaseModel): class ReleaseRunCommandRequest(BaseModel): request_id: str = Field(min_length=8, max_length=128) - reconciled: bool = False + + +class ReleaseRunReconcileRequest(BaseModel): + request_id: str = Field(min_length=8, max_length=128) + outcome: Literal["effect_absent", "effect_succeeded", "unresolved"] + confirm: str = Field(default="", max_length=32) def create_app( @@ -120,12 +127,13 @@ def create_app( app = FastAPI(title="GovOPlaN Release Console", version="0.1.0") app.state.workspace_root = workspace_root app.state.token = token + app.state.workspace_fingerprint = release_workspace_fingerprint(workspace_root) app.state.release_runs = ReleaseRunStore( run_state_root if run_state_root is not None - else default_release_run_root(workspace_root) + else default_release_run_root(workspace_root), + expected_workspace_fingerprint=app.state.workspace_fingerprint, ) - app.state.workspace_fingerprint = release_workspace_fingerprint(workspace_root) @app.middleware("http") async def require_token(request: Request, call_next): # type: ignore[no-untyped-def] @@ -236,7 +244,10 @@ def create_app( @app.get("/api/release-runs") def release_runs(limit: int = 20) -> dict[str, object]: - return {"runs": app.state.release_runs.list(limit=limit)} + try: + return {"runs": app.state.release_runs.list(limit=limit)} + except (ReleaseRunConflict, ReleaseRunCorrupt) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc @app.post("/api/release-runs") def create_release_run(request: ReleaseRunCreateRequest) -> dict[str, object]: @@ -303,7 +314,7 @@ def create_app( return app.state.release_runs.get(run_id) except ReleaseRunNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc - except ReleaseRunCorrupt as exc: + except (ReleaseRunConflict, ReleaseRunCorrupt) as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc @app.post("/api/release-runs/{run_id}/resume") @@ -326,7 +337,23 @@ def create_app( run_id, step_id, request_id=request.request_id, - reconciled=request.reconciled, + ) + except ReleaseRunNotFound as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except (ReleaseRunConflict, ReleaseRunCorrupt) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + @app.post("/api/release-runs/{run_id}/steps/{step_id}/reconcile") + def reconcile_release_run_step( + run_id: str, step_id: str, request: ReleaseRunReconcileRequest + ) -> dict[str, object]: + try: + return app.state.release_runs.reconcile_step( + run_id, + step_id, + request_id=request.request_id, + outcome=request.outcome, + confirmation=request.confirm, ) except ReleaseRunNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -451,14 +478,18 @@ def create_app( def default_release_run_root(workspace_root: Path) -> Path: - workspace = workspace_root.expanduser().resolve() - if (workspace / "repositories.json").is_file(): - return workspace / "runtime" / "release-runs" - workspace_meta = workspace / "govoplan" - if (workspace_meta / "repositories.json").is_file(): - return workspace_meta / "runtime" / "release-runs" - fingerprint = release_workspace_fingerprint(workspace) - return META_ROOT / "runtime" / "release-runs" / f"workspace-{fingerprint[:16]}" + configured_state_home = os.environ.get("XDG_STATE_HOME", "").strip() + state_home = Path(configured_state_home).expanduser() + if not configured_state_home or not state_home.is_absolute(): + state_home = Path.home() / ".local" / "state" + fingerprint = release_workspace_fingerprint(workspace_root) + return ( + state_home + / "govoplan" + / "release-console" + / f"workspace-{fingerprint}" + / "release-runs" + ) def release_workspace_fingerprint(workspace_root: Path) -> str: diff --git a/tools/release/webui/index.html b/tools/release/webui/index.html index f0c8131..c95e74e 100644 --- a/tools/release/webui/index.html +++ b/tools/release/webui/index.html @@ -907,6 +907,7 @@ manualRepo: "", currentRun: null, runs: [], + runStoreAvailable: true, }; elements.refresh.addEventListener("click", load); @@ -959,7 +960,11 @@ return fetch(`${path}${suffix}`, { headers: token ? { "X-Release-Console-Token": token } : {}, }).then((response) => { - if (!response.ok) throw new Error(`${response.status} ${response.statusText}`); + if (!response.ok) { + return response.json().catch(() => ({})).then((payload) => { + throw new Error(payload.detail || `${response.status} ${response.statusText}`); + }); + } return response.json(); }); } @@ -985,14 +990,21 @@ async function load() { elements.statusLine.textContent = "Refreshing..."; try { - const [dashboard, intelligence, runs] = await Promise.all([api("/api/dashboard"), api("/api/release-intelligence"), api("/api/release-runs")]); + const runsRequest = api("/api/release-runs") + .then((data) => ({ data, error: null })) + .catch((error) => ({ data: null, error })); + const [dashboard, intelligence, runsResult] = await Promise.all([api("/api/dashboard"), api("/api/release-intelligence"), runsRequest]); renderDashboard(dashboard); renderReleaseIntelligence(intelligence); - renderReleaseRunList(runs.runs || []); - if (releaseState.currentRun?.run_id) { - await loadReleaseRun(releaseState.currentRun.run_id, { restoreSelection: false }); + if (runsResult.error) { + renderReleaseRunStoreUnavailable(runsResult.error); } else { - renderIdlePlan(); + renderReleaseRunList(runsResult.data?.runs || []); + if (releaseState.currentRun?.run_id) { + await loadReleaseRun(releaseState.currentRun.run_id, { restoreSelection: false }); + } else { + renderIdlePlan(); + } } elements.statusLine.textContent = `Generated ${dashboard.generated_at}`; } catch (error) { @@ -1019,7 +1031,10 @@ } function renderReleaseRunList(runs) { + releaseState.runStoreAvailable = true; releaseState.runs = Array.isArray(runs) ? runs : []; + elements.releaseRun.disabled = false; + elements.createReleaseRun.disabled = false; const selectedId = releaseState.currentRun?.run_id || ""; const options = releaseState.runs.map((run) => { const repositories = run.repo_versions ? Object.keys(run.repo_versions).length : 0; @@ -1032,7 +1047,20 @@ } } + function renderReleaseRunStoreUnavailable(error) { + releaseState.runStoreAvailable = false; + releaseState.runs = []; + releaseState.currentRun = null; + elements.releaseRun.innerHTML = ``; + elements.releaseRun.disabled = true; + elements.createReleaseRun.disabled = true; + elements.resumeReleaseRun.disabled = true; + elements.runStatus.textContent = "unavailable"; + elements.runOutput.innerHTML = `

${pill("unavailable", "block")} Durable run storage cannot be read

${escapeHtml(error.message)}

Dashboard and release previews remain available. Correct the private state-directory problem, then refresh.

`; + } + async function createReleaseRun() { + if (!releaseState.runStoreAvailable) return; const repoVersions = selectedRepoVersions(); if (!Object.keys(repoVersions).length) { elements.runStatus.textContent = "selection required"; @@ -1110,7 +1138,7 @@ const recommendation = run.recommended_next || {}; const steps = Array.isArray(state.steps) ? state.steps : []; const events = Array.isArray(state.events) ? state.events.slice(-5).reverse() : []; - const recommendationKind = state.status === "blocked" ? "block" : recommendation.available ? "ok" : "warn"; + const recommendationKind = state.status === "blocked" ? "block" : state.status === "attention" ? "warn" : recommendation.available ? "ok" : "warn"; const recommendationHtml = recommendation.id ? `

${pill("recommended next", recommendationKind)} ${escapeHtml(recommendation.title || recommendation.id)}

${escapeHtml(recommendation.detail || "")}

@@ -1119,20 +1147,37 @@ const stepHtml = steps.map((step) => { const kind = step.state === "succeeded" ? "ok" : ["failed", "interrupted"].includes(step.state) ? "block" : step.available ? "ok" : "warn"; const retryReason = step.retry_available ? "Prepare this known failed or read-only interrupted step for an explicit retry." : (step.disabled_reason || "Retry is unavailable for this state."); + const reconciliationHtml = step.reconciliation_required ? `
+
+
+
+
` : ""; return `

${escapeHtml(`${step.order}. ${step.title || step.id}`)}

${pill(step.state, kind)}

${escapeHtml(step.detail || "")}

${step.result_code ? `

Result code: ${escapeHtml(step.result_code)}

` : ""}
+ ${reconciliationHtml}
`; }).join(""); const eventHtml = events.length ? `

Recent bounded state events

${events.map((event) => `

${escapeHtml(`${event.at} · ${event.type}${event.step_id ? ` · ${event.step_id}` : ""}${event.result_code ? ` · ${event.result_code}` : ""}`)}

`).join("")}
` : ""; elements.runStatus.textContent = state.status || "unknown"; - elements.resumeReleaseRun.disabled = state.status === "completed"; + elements.resumeReleaseRun.disabled = state.status !== "running"; elements.runOutput.innerHTML = `

${pill(state.status || "unknown", state.status === "blocked" ? "block" : state.status === "completed" ? "ok" : "warn")} ${escapeHtml(run.run_id)}

Frozen plan ${escapeHtml((run.immutable?.digest || "").slice(0, 16))}… · updated ${escapeHtml(run.updated_at || "-")}

Executor controls remain separate and keep their existing confirmations.

${recommendationHtml}${stepHtml}${eventHtml}`; for (const button of elements.runOutput.querySelectorAll("[data-run-retry-step]")) { button.addEventListener("click", () => retryReleaseRunStep(button.dataset.runRetryStep)); } + for (const controls of elements.runOutput.querySelectorAll("[data-run-reconcile-step]")) { + const outcome = controls.querySelector("[data-run-reconcile-outcome]"); + const confirmation = controls.querySelector("[data-run-reconcile-confirm]"); + const submit = controls.querySelector("[data-run-reconcile-submit]"); + const refresh = () => { + submit.disabled = !outcome.value || confirmation.value.trim() !== "RECONCILE"; + }; + outcome.addEventListener("change", refresh); + confirmation.addEventListener("input", refresh); + submit.addEventListener("click", () => reconcileReleaseRunStep(controls.dataset.runReconcileStep, outcome.value, confirmation.value.trim(), submit)); + } } async function resumeReleaseRun() { @@ -1147,7 +1192,7 @@ elements.runStatus.textContent = "error"; elements.runOutput.insertAdjacentHTML("afterbegin", `

${pill("error", "block")} Resume failed

${escapeHtml(error.message)}

`); } finally { - elements.resumeReleaseRun.disabled = releaseState.currentRun?.state?.status === "completed"; + elements.resumeReleaseRun.disabled = releaseState.currentRun?.state?.status !== "running"; } } @@ -1155,7 +1200,7 @@ const run = releaseState.currentRun; if (!run || !stepId) return; try { - const retried = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/retry`, { request_id: requestId("retry"), reconciled: false }); + const retried = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/retry`, { request_id: requestId("retry") }); releaseState.currentRun = retried; renderReleaseRun(retried); } catch (error) { @@ -1163,6 +1208,24 @@ } } + async function reconcileReleaseRunStep(stepId, outcome, confirmation, button) { + const run = releaseState.currentRun; + if (!run || !stepId || !outcome || confirmation !== "RECONCILE") return; + setBusy([button], true); + try { + const reconciled = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/reconcile`, { + request_id: requestId("reconcile"), + outcome, + confirm: confirmation, + }); + releaseState.currentRun = reconciled; + renderReleaseRun(reconciled); + } catch (error) { + elements.runOutput.insertAdjacentHTML("afterbegin", `

${pill("error", "block")} Reconciliation failed

${escapeHtml(error.message)}

`); + button.disabled = false; + } + } + function requestId(prefix) { const identifier = window.crypto?.randomUUID ? window.crypto.randomUUID() : `${Date.now()}-${Math.random().toString(16).slice(2)}`; return `${prefix}-${identifier}`;