feat(runtime): consume governed Activity Core closes
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a06ba0-10aa-7ea0-b20a-4f3fac39efe9
This commit is contained in:
parent
0e6d795aa4
commit
d00ffcb402
22 changed files with 1218 additions and 148 deletions
|
|
@ -17,12 +17,19 @@ import time
|
|||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from rein_aharness import metrics
|
||||
from rein_aharness.approaches import (
|
||||
APPROACH_UNMATCHED,
|
||||
ApproachResult,
|
||||
execute_approach,
|
||||
select_approach,
|
||||
)
|
||||
from rein_aharness.close_outbox import (
|
||||
CloseOutbox,
|
||||
CloseRequest,
|
||||
PermanentCloseDeliveryError,
|
||||
ReplayReport,
|
||||
)
|
||||
from rein_aharness.execution_cancel import (
|
||||
ExecutionCancel,
|
||||
ExecutionCancelled,
|
||||
|
|
@ -32,11 +39,13 @@ from rein_aharness.glas_execution import (
|
|||
GLAS_APPROACH,
|
||||
GlasExecutionError,
|
||||
execute_profiled_run,
|
||||
normalise_execution_evidence_for_close,
|
||||
)
|
||||
from rein_aharness.lease_monitor import LeaseLoss, LeaseMonitor
|
||||
from rein_aharness.repository_transaction import (
|
||||
DirtyRepositoryError,
|
||||
GitRepositoryError,
|
||||
RepositoryAcceptanceError,
|
||||
RepositoryBusyError,
|
||||
RepositoryTransaction,
|
||||
RepositoryTransactionError,
|
||||
|
|
@ -52,6 +61,17 @@ from rein_aharness.ops_run_client import (
|
|||
|
||||
logger = logging.getLogger("rein_aharness.claim_loop")
|
||||
|
||||
_PERMANENT_CLOSE_CODES = frozenset(
|
||||
{
|
||||
"not_found",
|
||||
"wrong_owner",
|
||||
"expired_lease",
|
||||
"state_conflict",
|
||||
"terminal_conflict",
|
||||
"evidence_conflict",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProcessResult:
|
||||
|
|
@ -180,16 +200,97 @@ def _cancelled_result(
|
|||
return None
|
||||
|
||||
|
||||
def _deliver_close(client: ActivityCoreOpsClient, request: CloseRequest) -> OpsRun:
|
||||
if request.worker_id != client.config.worker_id:
|
||||
raise PermanentCloseDeliveryError("outbox worker identity no longer matches")
|
||||
try:
|
||||
if request.action == "complete":
|
||||
return client.complete(request.run_id, result=request.result)
|
||||
return client.fail(
|
||||
request.run_id,
|
||||
error=request.error,
|
||||
reopen=request.reopen,
|
||||
result=request.result,
|
||||
)
|
||||
except OpsRunError as exc:
|
||||
if exc.code in _PERMANENT_CLOSE_CODES:
|
||||
raise PermanentCloseDeliveryError(
|
||||
f"Activity Core refused close with {exc.code}"
|
||||
) from exc
|
||||
raise
|
||||
|
||||
|
||||
def replay_close_outbox(
|
||||
client: ActivityCoreOpsClient,
|
||||
outbox: CloseOutbox | None = None,
|
||||
*,
|
||||
limit: int = 100,
|
||||
) -> ReplayReport:
|
||||
"""Replay close-only work; this function never dispatches a workload."""
|
||||
store = outbox or CloseOutbox()
|
||||
return store.replay(lambda request: _deliver_close(client, request), limit=limit)
|
||||
|
||||
|
||||
def _durable_close(
|
||||
client: ActivityCoreOpsClient,
|
||||
outbox: CloseOutbox,
|
||||
request: CloseRequest,
|
||||
) -> tuple[bool, str | None, str]:
|
||||
"""Enqueue before delivery and report the durable outcome for this entry."""
|
||||
receipt = outbox.enqueue(request)
|
||||
delivered: dict[str, OpsRun] = {}
|
||||
|
||||
def deliver(item: CloseRequest) -> None:
|
||||
delivered[item.entry_id] = _deliver_close(client, item)
|
||||
|
||||
report = outbox.replay(deliver)
|
||||
state = outbox.entry_state(receipt.entry_id)
|
||||
if state != "delivered":
|
||||
reason = (
|
||||
"close evidence quarantined"
|
||||
if state == "quarantined"
|
||||
else "close evidence remains pending"
|
||||
)
|
||||
return False, None, reason
|
||||
row = delivered.get(receipt.entry_id)
|
||||
if row is not None:
|
||||
return True, row.state, ""
|
||||
expected_state = "succeeded" if request.action == "complete" else "failed"
|
||||
if report.delivered:
|
||||
logger.info("replayed %s close evidence entries", report.delivered)
|
||||
return True, expected_state, ""
|
||||
|
||||
|
||||
def _bounded_reason(value: str, *, default: str) -> str:
|
||||
cleaned = " ".join(str(value).split())
|
||||
return cleaned[:500] or default
|
||||
|
||||
|
||||
def process_one(
|
||||
client: ActivityCoreOpsClient | None = None,
|
||||
*,
|
||||
report_to_hub: bool = True,
|
||||
commit: bool = True,
|
||||
dry_run: bool = False,
|
||||
outbox: CloseOutbox | None = None,
|
||||
) -> ProcessResult:
|
||||
"""Claim at most one ops_run, execute approach, complete or fail."""
|
||||
client = client or ActivityCoreOpsClient()
|
||||
cfg = client.config
|
||||
outbox = outbox or CloseOutbox()
|
||||
|
||||
replay = replay_close_outbox(client, outbox)
|
||||
if replay.remaining:
|
||||
return ProcessResult(
|
||||
claimed=False,
|
||||
retry_full_interval=True,
|
||||
reason=(
|
||||
"required Activity Core close evidence remains pending "
|
||||
f"(attempted={replay.attempted} remaining={replay.remaining})"
|
||||
),
|
||||
)
|
||||
if replay.quarantined:
|
||||
logger.error("quarantined %s close evidence entries", replay.quarantined)
|
||||
|
||||
try:
|
||||
claimed = client.claim(limit=1)
|
||||
|
|
@ -244,6 +345,36 @@ def process_one(
|
|||
client,
|
||||
run,
|
||||
report_to_hub=report_to_hub,
|
||||
outbox=outbox,
|
||||
)
|
||||
|
||||
if run.repository_grant is not None:
|
||||
reason = (
|
||||
"refused: repository_grant requires an authoritative "
|
||||
"harness_profile_ref"
|
||||
)
|
||||
try:
|
||||
out = client.fail(
|
||||
run.id,
|
||||
error=reason,
|
||||
reopen=False,
|
||||
result={"ok": False, "approach": approach, "reason": reason},
|
||||
)
|
||||
except OpsRunError as exc:
|
||||
return ProcessResult(
|
||||
claimed=True,
|
||||
run_id=run.id,
|
||||
approach=approach,
|
||||
ok=False,
|
||||
reason=f"grant-route refusal close failed: {exc}",
|
||||
)
|
||||
return ProcessResult(
|
||||
claimed=True,
|
||||
run_id=run.id,
|
||||
approach=approach,
|
||||
ok=False,
|
||||
reason=reason,
|
||||
ops_state=out.state,
|
||||
)
|
||||
|
||||
cancel = ExecutionCancel()
|
||||
|
|
@ -340,61 +471,213 @@ def _process_profiled_run(
|
|||
run: OpsRun,
|
||||
*,
|
||||
report_to_hub: bool,
|
||||
outbox: CloseOutbox,
|
||||
) -> ProcessResult:
|
||||
"""Execute a profiled row without consulting or falling back to legacy routing."""
|
||||
"""Keep the claim heartbeat alive through evidence persistence and close."""
|
||||
cancel = ExecutionCancel()
|
||||
hb = _Heartbeat(client, run.id, client.config.lease_seconds, cancel=cancel)
|
||||
hb.start()
|
||||
execution_error: str | None = None
|
||||
execution_reason = ""
|
||||
execution_reopen = False
|
||||
gateway_result: dict[str, Any] | None = None
|
||||
tx_evidence: dict[str, Any] | None = None
|
||||
_set_active_run_cancel(cancel)
|
||||
try:
|
||||
try:
|
||||
target = resolve_ops_target(run, client.config)
|
||||
with using_cancel(cancel):
|
||||
with RepositoryTransaction(target, correlation_id=run.id) as tx:
|
||||
gateway_result = execute_profiled_run(
|
||||
run,
|
||||
client.config,
|
||||
report_to_hub=report_to_hub,
|
||||
cancel=cancel,
|
||||
)
|
||||
tx_evidence = tx.evidence()
|
||||
except ExecutionCancelled as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = str(exc)
|
||||
except DirtyRepositoryError as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = f"refused: {exc}"
|
||||
except RepositoryBusyError as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = f"refused: {exc}"
|
||||
execution_reopen = True
|
||||
except (GitRepositoryError, RepositoryTransactionError, TaskSpecError) as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = f"refused: {exc}"
|
||||
except GlasExecutionError as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = str(exc)
|
||||
return _process_profiled_run_active(
|
||||
client,
|
||||
run,
|
||||
report_to_hub=report_to_hub,
|
||||
outbox=outbox,
|
||||
cancel=cancel,
|
||||
hb=hb,
|
||||
)
|
||||
finally:
|
||||
hb.stop()
|
||||
_set_active_run_cancel(None)
|
||||
|
||||
|
||||
def _process_profiled_run_active(
|
||||
client: ActivityCoreOpsClient,
|
||||
run: OpsRun,
|
||||
*,
|
||||
report_to_hub: bool,
|
||||
outbox: CloseOutbox,
|
||||
cancel: ExecutionCancel,
|
||||
hb: _Heartbeat,
|
||||
) -> ProcessResult:
|
||||
"""Execute a profiled row without consulting or falling back to legacy routing."""
|
||||
execution_error: str | None = None
|
||||
execution_reason = ""
|
||||
execution_reopen = False
|
||||
gateway_result: dict[str, Any] | None = None
|
||||
safe_evidence: dict[str, Any] = {}
|
||||
execution_ok = False
|
||||
tx: RepositoryTransaction | None = None
|
||||
tx_evidence: dict[str, Any] | None = None
|
||||
try:
|
||||
target = resolve_ops_target(run, client.config)
|
||||
with using_cancel(cancel):
|
||||
with RepositoryTransaction(target, correlation_id=run.id) as active_tx:
|
||||
tx = active_tx
|
||||
gateway_result = execute_profiled_run(
|
||||
run,
|
||||
client.config,
|
||||
report_to_hub=report_to_hub,
|
||||
cancel=cancel,
|
||||
)
|
||||
safe_evidence = normalise_execution_evidence_for_close(
|
||||
gateway_result["evidence"]
|
||||
)
|
||||
execution_ok = gateway_result["ok"]
|
||||
execution_reason = _bounded_reason(
|
||||
str(
|
||||
safe_evidence.get("error")
|
||||
or safe_evidence.get("outcome")
|
||||
or ""
|
||||
),
|
||||
default="Glas execution failed",
|
||||
)
|
||||
if run.repository_grant is not None:
|
||||
try:
|
||||
active_tx.validate_acceptance(
|
||||
run.repository_grant.acceptance_policy()
|
||||
)
|
||||
except RepositoryAcceptanceError as exc:
|
||||
if execution_ok or exc.code != "head-unchanged":
|
||||
execution_ok = False
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = _bounded_reason(
|
||||
str(exc),
|
||||
default="repository acceptance failed",
|
||||
)
|
||||
except ExecutionCancelled as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = _bounded_reason(str(exc), default="execution cancelled")
|
||||
except DirtyRepositoryError as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = _bounded_reason(
|
||||
f"refused: {exc}", default="dirty repository refused"
|
||||
)
|
||||
except RepositoryBusyError as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = _bounded_reason(
|
||||
f"refused: {exc}", default="busy repository refused"
|
||||
)
|
||||
execution_reopen = True
|
||||
except (GitRepositoryError, RepositoryTransactionError, TaskSpecError) as exc:
|
||||
execution_error = type(exc).__name__
|
||||
execution_reason = _bounded_reason(
|
||||
f"refused: {exc}", default="repository transaction refused"
|
||||
)
|
||||
except GlasExecutionError:
|
||||
execution_error = "GlasExecutionError"
|
||||
execution_reason = "profiled execution failed (GlasExecutionError)"
|
||||
|
||||
if tx is not None and tx.baseline is not None:
|
||||
tx_evidence = tx.evidence()
|
||||
if run.repository_grant is not None:
|
||||
tx_evidence["repository_grant"] = run.repository_grant.evidence()
|
||||
|
||||
if run.repository_grant is not None and tx is not None:
|
||||
try:
|
||||
accepted = tx.acceptance
|
||||
commit_sha = safe_evidence.get("commit_sha")
|
||||
metrics.record_external_execution(
|
||||
tx.repo,
|
||||
"rein-aharness",
|
||||
success=(
|
||||
execution_error is None
|
||||
and execution_ok
|
||||
and not cancel.cancelled
|
||||
and not hb.monitor.lost
|
||||
),
|
||||
committed=bool(accepted or commit_sha),
|
||||
head_after=(
|
||||
accepted.head
|
||||
if accepted is not None
|
||||
else commit_sha if isinstance(commit_sha, str) else None
|
||||
),
|
||||
reason=execution_reason or None,
|
||||
metadata={
|
||||
"ops_run_id": run.id,
|
||||
"harness_profile_ref": run.harness_profile_ref,
|
||||
"repository_grant_id": run.repository_grant.grant_id,
|
||||
},
|
||||
session_id=tx.transaction_id,
|
||||
)
|
||||
assert tx_evidence is not None
|
||||
tx_evidence["metrics"] = {
|
||||
"storage": "external",
|
||||
"session_id": tx.transaction_id,
|
||||
"projection_ready": True,
|
||||
}
|
||||
except OSError:
|
||||
execution_ok = False
|
||||
execution_error = "ExternalMetricsError"
|
||||
execution_reason = "required external metrics persistence failed"
|
||||
|
||||
# Re-check cancellation after metrics persistence. The heartbeat remains
|
||||
# active through this point and through close delivery, so a late signal or
|
||||
# lease loss cannot turn a cancelled execution into a successful close.
|
||||
refused = _cancelled_result(run.id, GLAS_APPROACH, hb)
|
||||
if refused is not None:
|
||||
if tx_evidence is not None:
|
||||
refused.detail["repository_transaction"] = tx_evidence
|
||||
if execution_error == "ExternalMetricsError":
|
||||
refused.reason = (
|
||||
f"{refused.reason}; required external metrics persistence failed"
|
||||
)
|
||||
if hb.monitor.lost or tx is None or tx_evidence is None:
|
||||
return refused
|
||||
payload = {
|
||||
"ok": False,
|
||||
"approach": GLAS_APPROACH,
|
||||
"reason": refused.reason,
|
||||
"repository_transaction": tx_evidence,
|
||||
}
|
||||
try:
|
||||
delivered, state, close_reason = _durable_close(
|
||||
client,
|
||||
outbox,
|
||||
CloseRequest(
|
||||
run_id=run.id,
|
||||
transaction_id=tx.transaction_id,
|
||||
worker_id=client.config.worker_id,
|
||||
action="fail",
|
||||
result=payload,
|
||||
error=refused.reason,
|
||||
reopen=False,
|
||||
),
|
||||
)
|
||||
except (OSError, RuntimeError) as exc:
|
||||
refused.reason = f"close evidence failure ({type(exc).__name__})"
|
||||
return refused
|
||||
if delivered:
|
||||
refused.ops_state = state
|
||||
else:
|
||||
refused.reason = close_reason
|
||||
return refused
|
||||
|
||||
if execution_error is not None:
|
||||
reason = execution_reason or f"profiled execution failed ({execution_error})"
|
||||
ok = execution_error is None and execution_ok
|
||||
reason = execution_reason or (
|
||||
"profiled execution failed"
|
||||
if not ok
|
||||
else ""
|
||||
)
|
||||
payload: dict[str, Any] = {
|
||||
"ok": ok,
|
||||
"approach": GLAS_APPROACH,
|
||||
}
|
||||
if safe_evidence:
|
||||
payload["execution_evidence"] = safe_evidence
|
||||
if reason:
|
||||
payload["reason"] = reason
|
||||
if tx_evidence is not None:
|
||||
payload["repository_transaction"] = tx_evidence
|
||||
|
||||
if tx is None or tx_evidence is None:
|
||||
try:
|
||||
out = client.fail(
|
||||
run.id,
|
||||
error=reason,
|
||||
reopen=execution_reopen,
|
||||
result={"ok": False, "approach": GLAS_APPROACH, "reason": reason},
|
||||
result=payload,
|
||||
)
|
||||
except OpsRunError as close_exc:
|
||||
return ProcessResult(
|
||||
|
|
@ -403,6 +686,7 @@ def _process_profiled_run(
|
|||
approach=GLAS_APPROACH,
|
||||
ok=False,
|
||||
reason=f"close ops_run failed: {close_exc}; {reason}",
|
||||
detail=payload,
|
||||
)
|
||||
return ProcessResult(
|
||||
claimed=True,
|
||||
|
|
@ -411,34 +695,43 @@ def _process_profiled_run(
|
|||
ok=False,
|
||||
reason=reason,
|
||||
ops_state=out.state,
|
||||
detail=payload,
|
||||
)
|
||||
|
||||
assert gateway_result is not None
|
||||
evidence = gateway_result["evidence"]
|
||||
ok = gateway_result["ok"]
|
||||
reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed")
|
||||
detail = {"execution_evidence": evidence}
|
||||
if tx_evidence is not None:
|
||||
detail["repository_transaction"] = tx_evidence
|
||||
gateway_result = {**gateway_result, "repository_transaction": tx_evidence}
|
||||
action = "complete" if ok else "fail"
|
||||
try:
|
||||
if ok:
|
||||
out = client.complete(run.id, result=gateway_result)
|
||||
else:
|
||||
out = client.fail(
|
||||
run.id,
|
||||
error=reason,
|
||||
reopen=False,
|
||||
result=gateway_result,
|
||||
)
|
||||
except OpsRunError as exc:
|
||||
close_request = CloseRequest(
|
||||
run_id=run.id,
|
||||
transaction_id=tx.transaction_id,
|
||||
worker_id=client.config.worker_id,
|
||||
action=action,
|
||||
result=payload,
|
||||
error=reason if action == "fail" else "",
|
||||
reopen=False,
|
||||
)
|
||||
delivered, state, close_reason = _durable_close(
|
||||
client,
|
||||
outbox,
|
||||
close_request,
|
||||
)
|
||||
except (OSError, RuntimeError) as exc:
|
||||
return ProcessResult(
|
||||
claimed=True,
|
||||
run_id=run.id,
|
||||
approach=GLAS_APPROACH,
|
||||
ok=False,
|
||||
reason=f"close ops_run failed: {exc}; gateway_ok={ok} {reason}",
|
||||
detail=detail,
|
||||
reason=f"close evidence failure ({type(exc).__name__})",
|
||||
detail=payload,
|
||||
)
|
||||
|
||||
if not delivered:
|
||||
return ProcessResult(
|
||||
claimed=True,
|
||||
run_id=run.id,
|
||||
approach=GLAS_APPROACH,
|
||||
ok=False,
|
||||
reason=close_reason,
|
||||
detail=payload,
|
||||
)
|
||||
|
||||
return ProcessResult(
|
||||
|
|
@ -447,8 +740,8 @@ def _process_profiled_run(
|
|||
approach=GLAS_APPROACH,
|
||||
ok=ok,
|
||||
reason="" if ok else reason,
|
||||
ops_state=out.state,
|
||||
detail=detail,
|
||||
ops_state=state,
|
||||
detail=payload,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue