diff --git a/src/k1link/observatory/recorded_jobs.py b/src/k1link/observatory/recorded_jobs.py index 8d24fa1..d34931b 100644 --- a/src/k1link/observatory/recorded_jobs.py +++ b/src/k1link/observatory/recorded_jobs.py @@ -36,6 +36,9 @@ from k1link.artifacts import utc_now_iso OBSERVATORY_RECORDED_JOB_SCHEMA: Final = "missioncore.observatory-recorded-job/v1" OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA: Final = "missioncore.observatory-recorded-job-request/v1" OBSERVATORY_RECORDED_CLAIM_SCHEMA: Final = "missioncore.observatory-recorded-job-claim/v1" +OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA: Final = ( + "missioncore.observatory-recorded-job-reconciliation/v1" +) OBSERVATORY_LIVE_LEASE_SCHEMA: Final = "missioncore.observatory-live-k1-lease/v1" OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA: Final = "missioncore.observatory-live-k1-lease-request/v1" RECORDED_JOB_DATABASE_NAME: Final = "observatory-recorded-jobs.sqlite3" @@ -152,6 +155,29 @@ CREATE TABLE IF NOT EXISTS observatory_recorded_claim_receipts ( FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id) ); +CREATE TABLE IF NOT EXISTS observatory_recorded_reconciliations ( + reconciliation_id TEXT PRIMARY KEY, + request_sha256 TEXT NOT NULL, + receipt_sha256 TEXT NOT NULL UNIQUE, + job_id TEXT NOT NULL UNIQUE, + expected_claim_generation INTEGER NOT NULL + CHECK (expected_claim_generation > 0), + expected_terminal_code TEXT NOT NULL, + quarantined_terminal_message TEXT NOT NULL, + quarantined_at_utc TEXT NOT NULL, + quarantined_claim_token_sha256 TEXT, + resource_release_attestation_id TEXT NOT NULL, + resource_release_attestation_sha256 TEXT NOT NULL, + resource_release_operator_id TEXT NOT NULL, + resource_release_evidence_sha256 TEXT NOT NULL, + resource_release_attested_at_utc TEXT NOT NULL, + resources_released INTEGER NOT NULL CHECK (resources_released = 1), + failure_code TEXT NOT NULL, + reason TEXT NOT NULL, + reconciled_at_utc TEXT NOT NULL, + FOREIGN KEY (job_id) REFERENCES observatory_recorded_jobs(job_id) +); + CREATE TABLE IF NOT EXISTS observatory_live_leases ( lease_id TEXT PRIMARY KEY, trigger_id TEXT NOT NULL UNIQUE, @@ -629,6 +655,168 @@ class ObservatoryRecordedClaim: } +@dataclass(frozen=True, slots=True) +class ObservatoryRecordedResourceReleaseAttestation: + """Operator-owned evidence that a stale Worker no longer owns resources.""" + + attestation_id: str + operator_id: str + job_id: str + claim_generation: int + resources_released: bool + evidence_sha256: str + attested_at_utc: str + + def __post_init__(self) -> None: + _validate_pattern(self.attestation_id, _IDEMPOTENCY_KEY, "attestation id") + _validate_pattern(self.operator_id, _IDENTIFIER, "operator id") + _validate_pattern(self.job_id, _JOB_ID, "recorded job id") + _validate_positive_int(self.claim_generation, "claim generation") + if self.resources_released is not True: + raise ValueError("resource-release attestation must explicitly release resources") + _validate_digest(self.evidence_sha256, "resource-release evidence sha256") + _validate_timestamp(self.attested_at_utc, "resource-release attestation timestamp") + + @property + def attestation_sha256(self) -> str: + return _sha256(self.identity_document()) + + def identity_document(self) -> dict[str, object]: + return { + "schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA, + "attestation_id": self.attestation_id, + "operator_id": self.operator_id, + "job_id": self.job_id, + "claim_generation": self.claim_generation, + "resources_released": self.resources_released, + "evidence_sha256": self.evidence_sha256, + "attested_at_utc": self.attested_at_utc, + } + + def as_dict(self) -> dict[str, object]: + return { + **self.identity_document(), + "attestation_sha256": self.attestation_sha256, + } + + +@dataclass(frozen=True, slots=True) +class ObservatoryRecordedReconciliationRequest: + """Fail-closed operator request to resolve one exact quarantined generation.""" + + reconciliation_id: str + job_id: str + expected_claim_generation: int + expected_terminal_code: str + resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation + failure_code: str + reason: str + + def __post_init__(self) -> None: + _validate_pattern(self.reconciliation_id, _IDEMPOTENCY_KEY, "reconciliation id") + _validate_pattern(self.job_id, _JOB_ID, "recorded job id") + _validate_positive_int(self.expected_claim_generation, "expected claim generation") + _validate_pattern( + self.expected_terminal_code, + _IDENTIFIER, + "expected terminal code", + ) + _validate_pattern(self.failure_code, _IDENTIFIER, "reconciliation failure code") + _validate_text(self.reason, "reconciliation reason", max_length=1_000) + if self.resource_release_attestation.job_id != self.job_id: + raise ValueError("resource-release attestation is bound to another job") + if ( + self.resource_release_attestation.claim_generation + != self.expected_claim_generation + ): + raise ValueError("resource-release attestation is bound to another generation") + + @property + def request_sha256(self) -> str: + return _sha256( + { + "schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA, + "reconciliation_id": self.reconciliation_id, + "job_id": self.job_id, + "expected_claim_generation": self.expected_claim_generation, + "expected_terminal_code": self.expected_terminal_code, + "resource_release_attestation": self.resource_release_attestation.as_dict(), + "failure_code": self.failure_code, + "reason": self.reason, + } + ) + + +@dataclass(frozen=True, slots=True) +class ObservatoryRecordedReconciliationReceipt: + """Immutable audit receipt for an operator-resolved recorded-job quarantine.""" + + reconciliation_id: str + request_sha256: str + receipt_sha256: str + job_id: str + expected_claim_generation: int + expected_terminal_code: str + quarantined_terminal_message: str + quarantined_at_utc: str + quarantined_claim_token_sha256: str | None + resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation + failure_code: str + reason: str + reconciled_at_utc: str + + def __post_init__(self) -> None: + request = ObservatoryRecordedReconciliationRequest( + reconciliation_id=self.reconciliation_id, + job_id=self.job_id, + expected_claim_generation=self.expected_claim_generation, + expected_terminal_code=self.expected_terminal_code, + resource_release_attestation=self.resource_release_attestation, + failure_code=self.failure_code, + reason=self.reason, + ) + if self.request_sha256 != request.request_sha256: + raise ObservatoryRecordedQueueIntegrityError( + "stored reconciliation request identity changed" + ) + _validate_text( + self.quarantined_terminal_message, + "quarantined terminal message", + max_length=1_000, + ) + _validate_timestamp(self.quarantined_at_utc, "quarantine timestamp") + _validate_optional_pattern( + self.quarantined_claim_token_sha256, + _SHA256, + "quarantined claim token sha256", + ) + _validate_timestamp(self.reconciled_at_utc, "reconciliation timestamp") + expected_receipt_sha256 = _sha256(self.identity_document()) + if self.receipt_sha256 != expected_receipt_sha256: + raise ObservatoryRecordedQueueIntegrityError( + "stored reconciliation receipt identity changed" + ) + + def identity_document(self) -> dict[str, object]: + return _reconciliation_receipt_identity_document( + reconciliation_id=self.reconciliation_id, + request_sha256=self.request_sha256, + job_id=self.job_id, + expected_claim_generation=self.expected_claim_generation, + expected_terminal_code=self.expected_terminal_code, + quarantined_terminal_message=self.quarantined_terminal_message, + quarantined_at_utc=self.quarantined_at_utc, + quarantined_claim_token_sha256=self.quarantined_claim_token_sha256, + resource_release_attestation=self.resource_release_attestation, + failure_code=self.failure_code, + reason=self.reason, + reconciled_at_utc=self.reconciled_at_utc, + ) + + def as_dict(self) -> dict[str, object]: + return {**self.identity_document(), "receipt_sha256": self.receipt_sha256} + + @dataclass(frozen=True, slots=True) class ObservatoryLiveLeaseIntent: trigger_id: str @@ -1212,6 +1400,173 @@ class ObservatoryRecordedJobQueue: ) return tuple(self._get_job(connection, job_id) for job_id in recovered_ids) + def reconcile_failed( + self, + request: ObservatoryRecordedReconciliationRequest, + ) -> ObservatoryRecordedReconciliationReceipt: + """Close one exact quarantine only after explicit resource-release proof.""" + + with self._transaction() as connection: + existing = connection.execute( + "SELECT * FROM observatory_recorded_reconciliations " + "WHERE reconciliation_id = ?", + (request.reconciliation_id,), + ).fetchone() + if existing is not None: + receipt = _reconciliation_receipt_from_row(existing) + if receipt.request_sha256 != request.request_sha256: + raise ObservatoryRecordedQueueConflictError( + "reconciliation id is bound to another request" + ) + self._require_reconciled_job_matches_receipt(connection, receipt) + return receipt + + prior = connection.execute( + "SELECT reconciliation_id FROM observatory_recorded_reconciliations " + "WHERE job_id = ?", + (request.job_id,), + ).fetchone() + if prior is not None: + raise ObservatoryRecordedQueueConflictError( + "recorded job is bound to another reconciliation" + ) + + job = self._get_job(connection, request.job_id) + if job.state != "reconciliation-required": + raise ObservatoryRecordedQueueConflictError( + "recorded job is not awaiting reconciliation" + ) + if job.claim_generation != request.expected_claim_generation: + raise ObservatoryRecordedQueueConflictError( + "recorded-job reconciliation generation changed" + ) + if job.terminal_code != request.expected_terminal_code: + raise ObservatoryRecordedQueueConflictError( + "recorded-job reconciliation terminal code changed" + ) + if job.terminal_message is None: + raise ObservatoryRecordedQueueIntegrityError( + "quarantined recorded job has no terminal reason" + ) + + reconciled_at_utc = self._timestamp() + if _parse_timestamp( + request.resource_release_attestation.attested_at_utc, + "resource-release attestation timestamp", + ) > _parse_timestamp(reconciled_at_utc, "reconciliation timestamp"): + raise ObservatoryRecordedQueueConflictError( + "resource-release attestation is newer than reconciliation" + ) + receipt_document = _reconciliation_receipt_identity_document( + reconciliation_id=request.reconciliation_id, + request_sha256=request.request_sha256, + job_id=request.job_id, + expected_claim_generation=request.expected_claim_generation, + expected_terminal_code=request.expected_terminal_code, + quarantined_terminal_message=job.terminal_message, + quarantined_at_utc=job.updated_at_utc, + quarantined_claim_token_sha256=job.terminal_claim_token_sha256, + resource_release_attestation=request.resource_release_attestation, + failure_code=request.failure_code, + reason=request.reason, + reconciled_at_utc=reconciled_at_utc, + ) + receipt = ObservatoryRecordedReconciliationReceipt( + reconciliation_id=request.reconciliation_id, + request_sha256=request.request_sha256, + receipt_sha256=_sha256(receipt_document), + job_id=request.job_id, + expected_claim_generation=request.expected_claim_generation, + expected_terminal_code=request.expected_terminal_code, + quarantined_terminal_message=job.terminal_message, + quarantined_at_utc=job.updated_at_utc, + quarantined_claim_token_sha256=job.terminal_claim_token_sha256, + resource_release_attestation=request.resource_release_attestation, + failure_code=request.failure_code, + reason=request.reason, + reconciled_at_utc=reconciled_at_utc, + ) + self._require_capacity( + connection, + table="observatory_recorded_reconciliations", + limit=self._max_jobs, + label="recorded reconciliation", + ) + connection.execute( + "INSERT INTO observatory_recorded_reconciliations " + "(reconciliation_id, request_sha256, receipt_sha256, job_id, " + "expected_claim_generation, expected_terminal_code, " + "quarantined_terminal_message, quarantined_at_utc, " + "quarantined_claim_token_sha256, " + "resource_release_attestation_id, " + "resource_release_attestation_sha256, resource_release_operator_id, " + "resource_release_evidence_sha256, resource_release_attested_at_utc, " + "resources_released, failure_code, reason, reconciled_at_utc) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + receipt.reconciliation_id, + receipt.request_sha256, + receipt.receipt_sha256, + receipt.job_id, + receipt.expected_claim_generation, + receipt.expected_terminal_code, + receipt.quarantined_terminal_message, + receipt.quarantined_at_utc, + receipt.quarantined_claim_token_sha256, + receipt.resource_release_attestation.attestation_id, + receipt.resource_release_attestation.attestation_sha256, + receipt.resource_release_attestation.operator_id, + receipt.resource_release_attestation.evidence_sha256, + receipt.resource_release_attestation.attested_at_utc, + int(receipt.resource_release_attestation.resources_released), + receipt.failure_code, + receipt.reason, + receipt.reconciled_at_utc, + ), + ) + updated = connection.execute( + "UPDATE observatory_recorded_jobs SET state = 'failed', " + "preemption_requested = 0, result_id = NULL, result_sha256 = NULL, " + "terminal_code = ?, terminal_message = ?, " + "terminal_claim_token_sha256 = NULL, updated_at_utc = ? " + "WHERE job_id = ? AND state = 'reconciliation-required' " + "AND claim_generation = ? AND terminal_code = ?", + ( + request.failure_code, + request.reason, + reconciled_at_utc, + request.job_id, + request.expected_claim_generation, + request.expected_terminal_code, + ), + ) + if updated.rowcount != 1: + raise ObservatoryRecordedQueueConflictError( + "recorded-job reconciliation guard changed" + ) + self._require_reconciled_job_matches_receipt(connection, receipt) + return receipt + + def get_reconciliation( + self, + job_id: str, + ) -> ObservatoryRecordedReconciliationReceipt: + """Return the immutable operator receipt for one reconciled job.""" + + _validate_pattern(job_id, _JOB_ID, "recorded job id") + with self._read_connection() as connection: + row = connection.execute( + "SELECT * FROM observatory_recorded_reconciliations WHERE job_id = ?", + (job_id,), + ).fetchone() + if row is None: + raise ObservatoryRecordedQueueNotFoundError( + "recorded-job reconciliation does not exist" + ) + receipt = _reconciliation_receipt_from_row(row) + self._require_reconciled_job_matches_receipt(connection, receipt) + return receipt + def start(self, job_id: str, *, claim_token: str) -> ObservatoryRecordedJob: """Enter running state, or yield before execution when live has priority.""" @@ -1968,6 +2323,26 @@ class ObservatoryRecordedJobQueue: raise ObservatoryRecordedQueueNotFoundError(job_id) return _job_from_row(row) + def _require_reconciled_job_matches_receipt( + self, + connection: sqlite3.Connection, + receipt: ObservatoryRecordedReconciliationReceipt, + ) -> None: + job = self._get_job(connection, receipt.job_id) + if ( + job.state != "failed" + or job.claim_generation != receipt.expected_claim_generation + or job.result_id is not None + or job.result_sha256 is not None + or job.terminal_code != receipt.failure_code + or job.terminal_message != receipt.reason + or job.terminal_claim_token_sha256 is not None + or job.updated_at_utc != receipt.reconciled_at_utc + ): + raise ObservatoryRecordedQueueIntegrityError( + "reconciled recorded job differs from its audit receipt" + ) + def _get_live_lease( self, connection: sqlite3.Connection, lease_id: str ) -> ObservatoryLiveLease: @@ -2030,6 +2405,7 @@ class ObservatoryRecordedJobQueue: expected = { "observatory_recorded_jobs": 46, "observatory_recorded_claim_receipts": 6, + "observatory_recorded_reconciliations": 18, "observatory_live_leases": 13, "observatory_recorded_preemptions": 14, } @@ -2146,6 +2522,11 @@ class ObservatoryRecordedJobQueue: self._max_jobs, "recorded preemption", ), + ( + "observatory_recorded_reconciliations", + self._max_jobs, + "recorded reconciliation", + ), ): count = len( connection.execute(f"SELECT 1 FROM {table} LIMIT ?", (limit + 1,)).fetchall() @@ -2245,6 +2626,47 @@ def _cancellation_request_from_row( ) from exc +def _reconciliation_receipt_from_row( + row: sqlite3.Row, +) -> ObservatoryRecordedReconciliationReceipt: + try: + attestation = ObservatoryRecordedResourceReleaseAttestation( + attestation_id=row["resource_release_attestation_id"], + operator_id=row["resource_release_operator_id"], + job_id=row["job_id"], + claim_generation=row["expected_claim_generation"], + resources_released=bool(row["resources_released"]), + evidence_sha256=row["resource_release_evidence_sha256"], + attested_at_utc=row["resource_release_attested_at_utc"], + ) + if ( + attestation.attestation_sha256 + != row["resource_release_attestation_sha256"] + ): + raise ObservatoryRecordedQueueIntegrityError( + "stored resource-release attestation identity changed" + ) + return ObservatoryRecordedReconciliationReceipt( + reconciliation_id=row["reconciliation_id"], + request_sha256=row["request_sha256"], + receipt_sha256=row["receipt_sha256"], + job_id=row["job_id"], + expected_claim_generation=row["expected_claim_generation"], + expected_terminal_code=row["expected_terminal_code"], + quarantined_terminal_message=row["quarantined_terminal_message"], + quarantined_at_utc=row["quarantined_at_utc"], + quarantined_claim_token_sha256=row["quarantined_claim_token_sha256"], + resource_release_attestation=attestation, + failure_code=row["failure_code"], + reason=row["reason"], + reconciled_at_utc=row["reconciled_at_utc"], + ) + except (IndexError, KeyError, TypeError, ValueError) as exc: + raise ObservatoryRecordedQueueIntegrityError( + "stored recorded-job reconciliation is invalid" + ) from exc + + def _job_from_row(row: sqlite3.Row) -> ObservatoryRecordedJob: try: checkpoints = json.loads(row["allowed_checkpoints_json"]) @@ -2409,6 +2831,41 @@ def _live_lease_states() -> frozenset[str]: return frozenset({"pending", "active", "completed", "failed", "cancelled"}) +def _reconciliation_receipt_identity_document( + *, + reconciliation_id: str, + request_sha256: str, + job_id: str, + expected_claim_generation: int, + expected_terminal_code: str, + quarantined_terminal_message: str, + quarantined_at_utc: str, + quarantined_claim_token_sha256: str | None, + resource_release_attestation: ObservatoryRecordedResourceReleaseAttestation, + failure_code: str, + reason: str, + reconciled_at_utc: str, +) -> dict[str, object]: + return { + "schema_version": OBSERVATORY_RECORDED_RECONCILIATION_SCHEMA, + "reconciliation_id": reconciliation_id, + "request_sha256": request_sha256, + "job_id": job_id, + "expected_claim_generation": expected_claim_generation, + "expected_terminal_code": expected_terminal_code, + "quarantined_terminal_message": quarantined_terminal_message, + "quarantined_at_utc": quarantined_at_utc, + "quarantined_claim_token_sha256": quarantined_claim_token_sha256, + "resource_release_attestation": resource_release_attestation.as_dict(), + "outcome": { + "state": "failed", + "code": failure_code, + "reason": reason, + }, + "reconciled_at_utc": reconciled_at_utc, + } + + def _canonical_json_text(value: object) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) diff --git a/tests/test_observatory_recorded_jobs.py b/tests/test_observatory_recorded_jobs.py index 5bf7e05..46090ab 100644 --- a/tests/test_observatory_recorded_jobs.py +++ b/tests/test_observatory_recorded_jobs.py @@ -19,6 +19,8 @@ from k1link.observatory.recorded_jobs import ( ObservatoryRecordedQueueConflictError, ObservatoryRecordedQueueIntegrityError, ObservatoryRecordedQueueStaleClaimError, + ObservatoryRecordedReconciliationRequest, + ObservatoryRecordedResourceReleaseAttestation, RecordedRunDefinition, RecordedRunDefinitionRegistry, ) @@ -36,6 +38,7 @@ EXECUTOR_RELEASE_SHA = "3" * 64 EXECUTOR_IMAGE_SHA = "4" * 64 MODEL_MANIFEST_SHA = "5" * 64 RESOURCE_PROFILE_SHA = "6" * 64 +RESOURCE_RELEASE_EVIDENCE_SHA = "7" * 64 def _definitions() -> RecordedRunDefinitionRegistry: @@ -155,6 +158,45 @@ def _running_job( return running, claim +def _resource_release_attestation( + job_id: str, + claim_generation: int, + *, + resources_released: bool = True, +) -> ObservatoryRecordedResourceReleaseAttestation: + return ObservatoryRecordedResourceReleaseAttestation( + attestation_id="worker-006-release-proof-001", + operator_id="missioncore-operator", + job_id=job_id, + claim_generation=claim_generation, + resources_released=resources_released, + evidence_sha256=RESOURCE_RELEASE_EVIDENCE_SHA, + attested_at_utc=NOW, + ) + + +def _reconciliation_request( + job_id: str, + claim_generation: int, + *, + reconciliation_id: str = "worker-006-reconciliation-001", + expected_terminal_code: str = "claim-lease-expired", + reason: str = "Worker 006 release was independently verified.", +) -> ObservatoryRecordedReconciliationRequest: + return ObservatoryRecordedReconciliationRequest( + reconciliation_id=reconciliation_id, + job_id=job_id, + expected_claim_generation=claim_generation, + expected_terminal_code=expected_terminal_code, + resource_release_attestation=_resource_release_attestation( + job_id, + claim_generation, + ), + failure_code="operator-reconciled-failure", + reason=reason, + ) + + def test_submission_is_durable_exactly_idempotent_and_path_free(tmp_path: Path) -> None: queue = _queue(tmp_path) first, first_created = queue.submit(_intent()) @@ -470,6 +512,131 @@ def test_expired_running_claim_is_quarantined_and_stale_terminal_is_fenced( ) +def test_operator_reconciliation_requires_exact_generation_code_and_release_proof( + tmp_path: Path, +) -> None: + clock_value = [NOW] + queue = ObservatoryRecordedJobQueue( + tmp_path, + definitions=_definitions(), + clock=lambda: clock_value[0], + claim_lease_seconds=10, + ) + running, _claim = _running_job(queue) + clock_value[0] = "2026-08-30T21:00:10.000Z" + [quarantined] = queue.recover_stale_claims() + + with pytest.raises(ValueError, match="explicitly release resources"): + _resource_release_attestation( + quarantined.job_id, + quarantined.claim_generation, + resources_released=False, + ) + with pytest.raises(ObservatoryRecordedQueueConflictError, match="generation"): + queue.reconcile_failed( + _reconciliation_request( + quarantined.job_id, + quarantined.claim_generation + 1, + ) + ) + with pytest.raises(ObservatoryRecordedQueueConflictError, match="terminal code"): + queue.reconcile_failed( + _reconciliation_request( + quarantined.job_id, + quarantined.claim_generation, + expected_terminal_code="claim-lease-migration", + ) + ) + + unchanged = queue.get(running.job_id) + assert unchanged.state == "reconciliation-required" + assert unchanged.terminal_code == "claim-lease-expired" + + +def test_operator_reconciliation_is_durable_idempotent_and_unblocks_queue( + tmp_path: Path, +) -> None: + clock_value = [NOW] + queue = ObservatoryRecordedJobQueue( + tmp_path, + definitions=_definitions(), + clock=lambda: clock_value[0], + claim_lease_seconds=10, + ) + running, _claim = _running_job(queue) + next_job, _ = queue.submit( + _intent( + idempotency_key="recorded-request-002", + source_session_id="another-session", + ), + enqueue=True, + ) + clock_value[0] = "2026-08-30T21:00:10.000Z" + [quarantined] = queue.recover_stale_claims() + request = _reconciliation_request( + quarantined.job_id, + quarantined.claim_generation, + ) + + receipt = queue.reconcile_failed(request) + replayed = queue.reconcile_failed(request) + restored_queue = ObservatoryRecordedJobQueue( + tmp_path, + definitions=_definitions(), + clock=lambda: clock_value[0], + claim_lease_seconds=10, + ) + restored = restored_queue.reconcile_failed(request) + + assert replayed == receipt + assert restored == receipt + assert restored_queue.get_reconciliation(running.job_id) == receipt + assert receipt.resource_release_attestation.resources_released is True + assert receipt.resource_release_attestation.evidence_sha256 == ( + RESOURCE_RELEASE_EVIDENCE_SHA + ) + assert receipt.expected_terminal_code == "claim-lease-expired" + assert receipt.quarantined_terminal_message == ( + "Worker claim lease expired after execution started; " + "physical resource ownership requires reconciliation." + ) + assert receipt.quarantined_at_utc == "2026-08-30T21:00:10.000Z" + assert receipt.as_dict()["outcome"] == { + "state": "failed", + "code": "operator-reconciled-failure", + "reason": "Worker 006 release was independently verified.", + } + failed = restored_queue.get(running.job_id) + assert failed.state == "failed" + assert failed.terminal_code == "operator-reconciled-failure" + assert failed.terminal_message == "Worker 006 release was independently verified." + assert failed.terminal_claim_token_sha256 is None + + with pytest.raises(ObservatoryRecordedQueueConflictError, match="another request"): + restored_queue.reconcile_failed( + _reconciliation_request( + quarantined.job_id, + quarantined.claim_generation, + reason="Conflicting operator explanation.", + ) + ) + with pytest.raises(ObservatoryRecordedQueueConflictError, match="another reconciliation"): + restored_queue.reconcile_failed( + _reconciliation_request( + quarantined.job_id, + quarantined.claim_generation, + reconciliation_id="worker-006-reconciliation-002", + ) + ) + + replacement = restored_queue.claim_next( + claimant_id="recorded-worker", + claim_request_id="claim-after-operator-reconciliation", + ) + assert replacement is not None + assert replacement.job.job_id == next_job.job_id + + def test_legacy_sqlite_claim_schema_migrates_without_reusing_old_token( tmp_path: Path, ) -> None: