Handle lease loss during adapter errors
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
This commit is contained in:
parent
75f4914777
commit
ad4074c20a
3 changed files with 98 additions and 32 deletions
|
|
@ -168,14 +168,18 @@ def process_one(
|
||||||
|
|
||||||
hb = _Heartbeat(client, run.id, cfg.lease_seconds)
|
hb = _Heartbeat(client, run.id, cfg.lease_seconds)
|
||||||
hb.start()
|
hb.start()
|
||||||
|
execution_error: str | None = None
|
||||||
try:
|
try:
|
||||||
ar: ApproachResult = execute_approach(
|
try:
|
||||||
run,
|
ar: ApproachResult = execute_approach(
|
||||||
approach=approach,
|
run,
|
||||||
config=cfg,
|
approach=approach,
|
||||||
report_to_hub=report_to_hub,
|
config=cfg,
|
||||||
commit=commit,
|
report_to_hub=report_to_hub,
|
||||||
)
|
commit=commit,
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
execution_error = type(exc).__name__
|
||||||
finally:
|
finally:
|
||||||
hb.stop()
|
hb.stop()
|
||||||
|
|
||||||
|
|
@ -196,6 +200,16 @@ def process_one(
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if execution_error is not None:
|
||||||
|
return ProcessResult(
|
||||||
|
claimed=True,
|
||||||
|
run_id=run.id,
|
||||||
|
approach=approach,
|
||||||
|
ok=False,
|
||||||
|
reason=f"approach failed ({execution_error})",
|
||||||
|
detail={"execution_error_type": execution_error},
|
||||||
|
)
|
||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"approach": ar.approach,
|
"approach": ar.approach,
|
||||||
"ok": ar.ok,
|
"ok": ar.ok,
|
||||||
|
|
@ -259,6 +273,9 @@ def _process_profiled_run(
|
||||||
"""Execute a profiled row without consulting or falling back to legacy routing."""
|
"""Execute a profiled row without consulting or falling back to legacy routing."""
|
||||||
hb = _Heartbeat(client, run.id, client.config.lease_seconds)
|
hb = _Heartbeat(client, run.id, client.config.lease_seconds)
|
||||||
hb.start()
|
hb.start()
|
||||||
|
execution_error: str | None = None
|
||||||
|
execution_reason = ""
|
||||||
|
gateway_result: dict[str, Any] | None = None
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
gateway_result = execute_profiled_run(
|
gateway_result = execute_profiled_run(
|
||||||
|
|
@ -267,30 +284,8 @@ def _process_profiled_run(
|
||||||
report_to_hub=report_to_hub,
|
report_to_hub=report_to_hub,
|
||||||
)
|
)
|
||||||
except GlasExecutionError as exc:
|
except GlasExecutionError as exc:
|
||||||
reason = str(exc)
|
execution_error = type(exc).__name__
|
||||||
try:
|
execution_reason = str(exc)
|
||||||
out = client.fail(
|
|
||||||
run.id,
|
|
||||||
error=reason,
|
|
||||||
reopen=False,
|
|
||||||
result={"ok": False, "approach": GLAS_APPROACH, "reason": reason},
|
|
||||||
)
|
|
||||||
except OpsRunError as close_exc:
|
|
||||||
return ProcessResult(
|
|
||||||
claimed=True,
|
|
||||||
run_id=run.id,
|
|
||||||
approach=GLAS_APPROACH,
|
|
||||||
ok=False,
|
|
||||||
reason=f"close ops_run failed: {close_exc}; {reason}",
|
|
||||||
)
|
|
||||||
return ProcessResult(
|
|
||||||
claimed=True,
|
|
||||||
run_id=run.id,
|
|
||||||
approach=GLAS_APPROACH,
|
|
||||||
ok=False,
|
|
||||||
reason=reason,
|
|
||||||
ops_state=out.state,
|
|
||||||
)
|
|
||||||
finally:
|
finally:
|
||||||
hb.stop()
|
hb.stop()
|
||||||
|
|
||||||
|
|
@ -311,6 +306,33 @@ def _process_profiled_run(
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if execution_error is not None:
|
||||||
|
reason = execution_reason or f"profiled execution failed ({execution_error})"
|
||||||
|
try:
|
||||||
|
out = client.fail(
|
||||||
|
run.id,
|
||||||
|
error=reason,
|
||||||
|
reopen=False,
|
||||||
|
result={"ok": False, "approach": GLAS_APPROACH, "reason": reason},
|
||||||
|
)
|
||||||
|
except OpsRunError as close_exc:
|
||||||
|
return ProcessResult(
|
||||||
|
claimed=True,
|
||||||
|
run_id=run.id,
|
||||||
|
approach=GLAS_APPROACH,
|
||||||
|
ok=False,
|
||||||
|
reason=f"close ops_run failed: {close_exc}; {reason}",
|
||||||
|
)
|
||||||
|
return ProcessResult(
|
||||||
|
claimed=True,
|
||||||
|
run_id=run.id,
|
||||||
|
approach=GLAS_APPROACH,
|
||||||
|
ok=False,
|
||||||
|
reason=reason,
|
||||||
|
ops_state=out.state,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert gateway_result is not None
|
||||||
evidence = gateway_result["evidence"]
|
evidence = gateway_result["evidence"]
|
||||||
ok = gateway_result["ok"]
|
ok = gateway_result["ok"]
|
||||||
reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed")
|
reason = str(evidence.get("error") or evidence.get("outcome") or "Glas execution failed")
|
||||||
|
|
|
||||||
|
|
@ -135,6 +135,48 @@ def test_process_one_refuses_close_after_lease_loss() -> None:
|
||||||
client.fail.assert_not_called()
|
client.fail.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_process_one_records_adapter_exception_without_unbound_result() -> None:
|
||||||
|
client = MagicMock(spec=ActivityCoreOpsClient)
|
||||||
|
client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
|
||||||
|
client.claim.return_value = [_claimed_run()]
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"rein_aharness.claim_loop.execute_approach",
|
||||||
|
side_effect=RuntimeError("provider output must not be retained"),
|
||||||
|
):
|
||||||
|
result = process_one(client)
|
||||||
|
|
||||||
|
assert result.ok is False
|
||||||
|
assert result.reason == "approach failed (RuntimeError)"
|
||||||
|
assert result.detail == {"execution_error_type": "RuntimeError"}
|
||||||
|
client.complete.assert_not_called()
|
||||||
|
client.fail.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
def test_profiled_exception_after_lease_loss_skips_close() -> None:
|
||||||
|
client = MagicMock(spec=ActivityCoreOpsClient)
|
||||||
|
client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
|
||||||
|
run = _claimed_run()
|
||||||
|
run.harness_profile_ref = "harness.agent-dev-local@1.0.0"
|
||||||
|
client.claim.return_value = [run]
|
||||||
|
client.heartbeat.side_effect = OpsRunError("lease rejected")
|
||||||
|
|
||||||
|
def slow_profile(*_args, **_kwargs):
|
||||||
|
time.sleep(0.05)
|
||||||
|
raise GlasExecutionError("profile failed")
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01),
|
||||||
|
patch("rein_aharness.claim_loop.execute_profiled_run", side_effect=slow_profile),
|
||||||
|
):
|
||||||
|
result = process_one(client)
|
||||||
|
|
||||||
|
assert result.ok is False
|
||||||
|
assert result.reason.startswith("lease lost")
|
||||||
|
client.complete.assert_not_called()
|
||||||
|
client.fail.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
def test_poll_peek() -> None:
|
def test_poll_peek() -> None:
|
||||||
client = MagicMock(spec=ActivityCoreOpsClient)
|
client = MagicMock(spec=ActivityCoreOpsClient)
|
||||||
client.list_open.return_value = [_claimed_run()]
|
client.list_open.return_value = [_claimed_run()]
|
||||||
|
|
|
||||||
|
|
@ -204,7 +204,9 @@ responses, connect cancellation at each adapter boundary, and refuse result
|
||||||
acceptance/close after loss. The claim loop now implements that last refusal for
|
acceptance/close after loss. The claim loop now implements that last refusal for
|
||||||
normal and successful profiled executions: a lost lease returns bounded loss
|
normal and successful profiled executions: a lost lease returns bounded loss
|
||||||
evidence and skips Activity Core completion/failure calls, leaving reconciliation
|
evidence and skips Activity Core completion/failure calls, leaving reconciliation
|
||||||
to the owner of the expired lease.
|
to the owner of the expired lease. Adapter exception paths now also stop before
|
||||||
|
close when lease loss raced the failure; otherwise they return only an exception
|
||||||
|
class marker rather than leaving an unbound result or retaining provider text.
|
||||||
|
|
||||||
## Verify accepted commits and reconcile metrics/reporting
|
## Verify accepted commits and reconcile metrics/reporting
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue