Skip to content

fix: expose accepted execution backend in heartbeat output #502

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Sep 9, 2026
Merged

fix: expose accepted execution backend in heartbeat output #502

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 23 additions & 7 deletions scripts/execution_report_heartbeat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,21 @@
DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"}
DEFAULT_REJECT_STATUSES = {"error", "failed", "failure", "cancelled", "canceled", "timed_out"}
DEFAULT_REJECT_EXECUTION_STATUSES = {"error", "failed", "failure"}
REPORTABLE_EXECUTION_BACKENDS = frozenset({"gateway", "quantconnect"})
EXPECTED_EXECUTION_GUARD_REASON_PREFIXES = (
"pending_orders_detected:",
"same_day_fills_detected:",
"same_day_execution_locked:",
)


def _report_execution_backend(payload: Mapping[str, Any]) -> str:
backend = payload.get("execution_backend")
if isinstance(backend, str) and backend in REPORTABLE_EXECUTION_BACKENDS:
return backend
return "unknown"


DEFAULT_ACCEPT_STAGES = {
"DRY_RUN_COMPLETED",
"FUNDING_BLOCKED",
Expand Down Expand Up @@ -1146,8 +1156,8 @@ def main(now: dt.datetime | None = None) -> int:
list_errors.append(f"{gcs_glob}: {exc}")

sorted_objects = sorted(objects.values(), key=lambda item: item[1], reverse=True)
accepted = []
accepted_by_service: dict[str, tuple[str, dt.datetime, str]] = {}
accepted: list[tuple[str, dt.datetime, str, str]] = []
accepted_by_service: dict[str, tuple[str, dt.datetime, str, str]] = {}
inspected = []
for uri, updated in sorted_objects[:max_reports]:
payload = _cat_gcs_json(uri, project=project)
Expand All @@ -1172,24 +1182,30 @@ def main(now: dt.datetime | None = None) -> int:
ok, reason = _is_accepted_report(payload)
inspected.append(f"- {updated.isoformat()} {uri} {reason}")
if ok:
execution_backend = _report_execution_backend(payload)
if required_keys:
accepted_by_service[service_name] = (uri, updated, reason)
accepted_by_service.setdefault(
service_name,
(uri, updated, reason, execution_backend),
)
else:
accepted.append((uri, updated, reason))
accepted.append((uri, updated, reason, execution_backend))

if required_keys:
missing = [key for key in required_keys if key not in accepted_by_service]
if not missing:
details = ", ".join(
f"{required_labels[key]}@{accepted_by_service[key][1].isoformat()}"
f"{required_labels[key]}@{accepted_by_service[key][1].isoformat()} "
f"backend={accepted_by_service[key][3]}"
for key in required_keys
)
print(f"Execution report heartbeat OK for {name}: {details}")
return 0
if accepted:
uri, updated, reason = accepted[0]
uri, updated, reason, execution_backend = accepted[0]
print(
f"Execution report heartbeat OK for {name}: {reason}, updated={updated.isoformat()}, uri={uri}"
f"Execution report heartbeat OK for {name}: {reason}, updated={updated.isoformat()}, "
f"backend={execution_backend}, uri={uri}"
)
return 0

Expand Down
152 changes: 152 additions & 0 deletions tests/test_execution_report_heartbeat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -584,6 +584,158 @@ def test_main_rejects_previous_session_report_after_a_new_session_is_due(
assert "predates latest due schedule" in output


def test_main_reports_backend_from_newest_accepted_required_target_report(
monkeypatch,
capsys,
):
_clear_runtime_env(monkeypatch)
monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "IBKR runtime")
monkeypatch.setenv("RUNTIME_HEARTBEAT_REQUIRED_SERVICES", "svc-us")
monkeypatch.setenv("RUNTIME_HEARTBEAT_GCS_URIS", "gs://bucket/reports")
monkeypatch.setenv(
"CLOUD_RUN_SERVICE_TARGETS_JSON",
json.dumps(
{
"targets": [
{
"service": "svc-us",
"runtime_target": {
"service_name": "svc-us",
"strategy_profile": "us-strategy",
"account_scope": "US",
"scheduler": {
"timezone": "UTC",
"main_time": "0 8 * * *",
},
},
}
]
}
),
)
monkeypatch.setattr(
heartbeat,
"_list_gcs_objects",
lambda *_args, **_kwargs: [
{
"url": "gs://bucket/reports/newer.json",
"metadata": {"updated": "2026-09-09T10:00:00Z"},
},
{
"url": "gs://bucket/reports/older.json",
"metadata": {"updated": "2026-09-09T09:00:00Z"},
},
],
)
monkeypatch.setattr(
heartbeat,
"_cat_gcs_json",
lambda uri, **_kwargs: {
"status": "ok",
"service_name": "svc-us",
"strategy_profile": "us-strategy",
"account_scope": "US",
"execution_backend": "gateway" if uri.endswith("newer.json") else "quantconnect",
},
)

result = heartbeat.main(now=dt.datetime(2026, 9, 9, 10, 30, tzinfo=dt.timezone.utc))

assert result == 0
output = capsys.readouterr().out
assert "svc-us[us-strategy/US]@2026-09-09T10:00:00+00:00 backend=gateway" in output
assert "quantconnect" not in output


@pytest.mark.parametrize(
"execution_backend",
[None, 1, "gateway;secret=not-a-backend", "GATEWAY"],
)
def test_main_reports_unknown_backend_for_noncanonical_no_required_report(
monkeypatch,
capsys,
execution_backend,
):
_clear_runtime_env(monkeypatch)
monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "IBKR runtime")
monkeypatch.setenv("RUNTIME_HEARTBEAT_GCS_URIS", "gs://bucket/reports")
monkeypatch.setattr(
heartbeat,
"_list_gcs_objects",
lambda *_args, **_kwargs: [
{
"url": "gs://bucket/reports/accepted.json",
"metadata": {"updated": "2026-09-09T10:00:00Z"},
}
],
)
monkeypatch.setattr(
heartbeat,
"_cat_gcs_json",
lambda *_args, **_kwargs: {
"status": "ok",
"service_name": "svc-us",
"execution_backend": execution_backend,
},
)

result = heartbeat.main(now=dt.datetime(2026, 9, 9, 10, 30, tzinfo=dt.timezone.utc))

assert result == 0
output = capsys.readouterr().out
assert "backend=unknown" in output
if isinstance(execution_backend, str):
assert execution_backend not in output


def test_main_does_not_use_rejected_report_backend_for_no_required_report(
monkeypatch,
capsys,
):
_clear_runtime_env(monkeypatch)
monkeypatch.setenv("RUNTIME_HEARTBEAT_NAME", "IBKR runtime")
monkeypatch.setenv("RUNTIME_HEARTBEAT_GCS_URIS", "gs://bucket/reports")
monkeypatch.setattr(
heartbeat,
"_list_gcs_objects",
lambda *_args, **_kwargs: [
{
"url": "gs://bucket/reports/rejected.json",
"metadata": {"updated": "2026-09-09T10:00:00Z"},
},
{
"url": "gs://bucket/reports/accepted.json",
"metadata": {"updated": "2026-09-09T09:00:00Z"},
},
],
)
monkeypatch.setattr(
heartbeat,
"_cat_gcs_json",
lambda uri, **_kwargs: (
{
"status": "ok",
"service_name": "svc-us",
"execution_backend": "gateway",
"summary": {"execution_status": "blocked", "no_op_reason": "no_equity"},
}
if uri.endswith("rejected.json")
else {
"status": "ok",
"service_name": "svc-us",
"execution_backend": "quantconnect",
}
),
)

result = heartbeat.main(now=dt.datetime(2026, 9, 9, 10, 30, tzinfo=dt.timezone.utc))

assert result == 0
output = capsys.readouterr().out
assert "backend=quantconnect" in output
assert "backend=gateway" not in output


def test_report_with_blocked_execution_status_is_rejected_even_when_top_level_is_ok():
accepted, reason = heartbeat._is_accepted_report(
{
Expand Down