diff --git a/docs/edge-relay-resilience.md b/docs/edge-relay-resilience.md new file mode 100644 index 0000000..992828d --- /dev/null +++ b/docs/edge-relay-resilience.md @@ -0,0 +1,77 @@ +# Edge relay resilience (ACTIVITY-WP-0027-T06) + +**Audience:** operators and agents debugging high-frequency activity-core jobs +**Scope:** railiance01 `actcore-statehub-edge-relay` + worker `STATE_HUB_URL` + +## Topology + +```text +actcore-worker + STATE_HUB_URL=http://actcore-statehub-edge-relay:8000 + │ + ▼ + edge relay (cluster) + · GET allowlist → upstream or stale read-cache + · POST side-effects → upstream (or 503 if busy/unreachable) + │ + ▼ + state-hub service (cluster DNS state-hub.state-hub.svc …) + · may depend on tunnel / central hub posture +``` + +Host rein-aharness should prefer the **same edge ClusterIP** (or `k8s://…edge-relay:8000`) +so completion events outbox when central hub is flaky — see +`docs/llm-connect-host-access.md`. + +## Resolver classes + +| Class | Examples | Hard-fail? | Notes | +| ----- | -------- | ---------- | ----- | +| **GET reads** | `fi_brief_status`, `domain_summary`, `state_summary` | Soft by default (`{}` / empty) unless `required: true` | Edge may serve **stale cache** (`X-StateHub-Edge-Cache: stale`) | +| **Side-effect POSTs** | `consistency_sweep_remote_all`, `recently_on_scope_hourly` | After retries: **degrade** by default | See env below | +| **Required pure reads** | `todo_md_staleness` when enabled | Yes if `required: true` | Prefer not to make hub-only reads required without cache | + +## Side-effect POST behaviour (worker) + +Implemented in `context_resolvers/state_hub.py` (`_post_json`): + +1. **Retry** transient `502/503/504` and network/timeout errors + (`STATE_HUB_POST_RETRIES`, default **3**; backoff + `STATE_HUB_POST_RETRY_BACKOFF_SECONDS`, default **2s** × attempt). +2. If still failing and degrade is on (`STATE_HUB_SIDE_EFFECT_DEGRADE` default + **true**, or per-source `params.degrade_on_unavailable`): + - **consistency_sweep:** returns + `{exit_code: 75, lock_skipped: true, repos_processed: [], degraded: true, …}` + so the workflow **completes** without Temporal ApplicationError storms. + - **recently_on_scope:** returns + `{generated: [], failed: [{reason: edge_unavailable}], degraded: true, …}`. +3. Set `degrade_on_unavailable: false` on a definition (or env + `STATE_HUB_SIDE_EFFECT_DEGRADE=false`) to keep hard-fail + Temporal retry + after in-resolver attempts. + +## Operator checks + +```bash +# Edge health (from host with ClusterIP or kubectl exec) +curl -sS http://:8000/edge/health | python3 -m json.tool + +# Worker still pointing at edge +kubectl -n activity-core exec deploy/actcore-worker -- env | grep STATE_HUB + +# Recent consistency fires should COMPLETE even when edge was briefly 503 +# (look for degraded:true in context_snapshot if all retries failed) +``` + +## What edge 503 usually means + +- Upstream hub slow / lock held / temporary overload on + `/consistency/sweep/remote-all` (large fleet). +- Upstream unreachable; edge refuses to invent a false success for POSTs. +- Outbox backlog (see `edge/health` → `outbox.pending_count`) is **write** + backlog for queued progress — separate from sweep POST 503. + +## Non-goals + +- Making the edge invent successful sweep results when nothing ran. +- Public exposure of State Hub or edge. +- Replacing Temporal schedules with workstation heartbeats. diff --git a/docs/recurring-automations-playbook.md b/docs/recurring-automations-playbook.md index 8f549fa..3515c00 100644 --- a/docs/recurring-automations-playbook.md +++ b/docs/recurring-automations-playbook.md @@ -50,7 +50,9 @@ activity-core also schedules it — dual cadence and forgotten design. **Primary (Binky / FI):** activity-core schedule → `ops_run` → rein-aharness **claim loop** on railiance01. Host user-systemd timers are **break-glass only** (not cadence authority). FI dual-clock timer was disabled 2026-08-05 after the -ops_run per-fire idempotency fix landed; re-enable only if claim loop is down. +ops_run per-fire idempotency fix; **Binky dual-clock timers** (daily/mail/review) +were disabled 2026-08-05 after claim-loop smoke (ACTIVITY-WP-0027-T07). Re-enable +only if claim loop is down. --- diff --git a/docs/runbook.md b/docs/runbook.md index ff0f187..9f1f518 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -736,6 +736,11 @@ unreachable, which keeps daily triage context resolution alive during outages. > receipts as successful sink delivery pending replay, and consumes stale cached > reads transparently. +Side-effect POSTs (`consistency_sweep_remote_all`, `recently_on_scope_hourly`) +retry transient 502/503/504 then **degrade** by default so required workflows do +not thrash Temporal when the edge is briefly unavailable. Details: +`docs/edge-relay-resilience.md` (ACTIVITY-WP-0027-T06). + ## Troubleshooting ### Worker fails to start: "ACTCORE_DB_URL is required" diff --git a/src/activity_core/context_resolvers/state_hub.py b/src/activity_core/context_resolvers/state_hub.py index 72531d1..4cc7a11 100644 --- a/src/activity_core/context_resolvers/state_hub.py +++ b/src/activity_core/context_resolvers/state_hub.py @@ -26,13 +26,21 @@ When STATE_HUB_URL points at the state-hub edge relay, allowlisted GET reads may be served from a stale local cache during upstream outages (`X-StateHub-Edge-Cache: stale`). activity-core treats those as ordinary successful reads so workflows can continue with last-known hub state. + +Side-effect POSTs (`consistency_sweep_remote_all`, `recently_on_scope_hourly`) +retry transient 502/503/504 and network errors, then optionally **degrade** +instead of hard-failing the workflow (ACTIVITY-WP-0027-T06). See +`docs/edge-relay-resilience.md`. + Config: STATE_HUB_URL env var (default: http://127.0.0.1:8000). """ from __future__ import annotations import json +import logging import os +import time from datetime import datetime, timezone from typing import Any @@ -40,6 +48,8 @@ import httpx from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver +logger = logging.getLogger(__name__) + _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000" _TIMEOUT_SECONDS = 10.0 _SWEEP_TIMEOUT_SECONDS = 330.0 @@ -50,6 +60,9 @@ _OPEN_TASK_STATUSES = {"wait", "todo", "progress"} # forcing the rule expression to special-case None. _NEVER_SCANNED_AGE_DAYS = 99999 +# Transient HTTP statuses worth retrying against the edge relay / hub. +_TRANSIENT_HTTP = frozenset({502, 503, 504}) + def _base_url() -> str: return os.environ.get("STATE_HUB_URL", _DEFAULT_STATE_HUB_URL).rstrip("/") @@ -65,11 +78,92 @@ def _fetch_json(path: str, params: dict[str, Any] | None = None) -> Any: return {} -def _post_json(path: str, payload: dict[str, Any], *, timeout: float = _TIMEOUT_SECONDS) -> Any: +def _post_retry_attempts() -> int: + try: + return max(1, int(os.environ.get("STATE_HUB_POST_RETRIES", "3"))) + except ValueError: + return 3 + + +def _post_retry_backoff_seconds() -> float: + try: + return max(0.0, float(os.environ.get("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "2"))) + except ValueError: + return 2.0 + + +def _side_effect_degrade_default() -> bool: + """Default degrade-on-unavailable for side-effect POSTs after retries.""" + raw = (os.environ.get("STATE_HUB_SIDE_EFFECT_DEGRADE") or "true").strip().lower() + return raw in {"1", "true", "yes", "on", ""} + + +def _post_json( + path: str, + payload: dict[str, Any], + *, + timeout: float = _TIMEOUT_SECONDS, + retries: int | None = None, +) -> Any: + """POST JSON with retries on transient edge/hub failures (ACTIVITY-WP-0027).""" url = f"{_base_url()}{path}" - resp = httpx.post(url, json=payload, timeout=timeout) - resp.raise_for_status() - return resp.json() + attempts = _post_retry_attempts() if retries is None else max(1, retries) + backoff = _post_retry_backoff_seconds() + last_exc: Exception | None = None + + for attempt in range(attempts): + try: + resp = httpx.post(url, json=payload, timeout=timeout) + status = getattr(resp, "status_code", None) + if status in _TRANSIENT_HTTP and attempt + 1 < attempts: + logger.warning( + "state-hub POST %s returned %s (attempt %s/%s); retrying", + path, + status, + attempt + 1, + attempts, + ) + if backoff: + time.sleep(backoff * (attempt + 1)) + continue + resp.raise_for_status() + return resp.json() + except ( + httpx.TimeoutException, + httpx.NetworkError, + httpx.RemoteProtocolError, + ) as exc: + last_exc = exc + if attempt + 1 >= attempts: + raise + logger.warning( + "state-hub POST %s network error %s (attempt %s/%s); retrying", + path, + exc, + attempt + 1, + attempts, + ) + if backoff: + time.sleep(backoff * (attempt + 1)) + except httpx.HTTPStatusError as exc: + # Non-transient HTTP errors fail immediately. + code = exc.response.status_code if exc.response is not None else None + if code in _TRANSIENT_HTTP and attempt + 1 < attempts: + last_exc = exc + if backoff: + time.sleep(backoff * (attempt + 1)) + continue + raise + + if last_exc is not None: + raise last_exc + raise RuntimeError(f"state-hub POST {path} failed without exception") + + +def _want_degrade(params: dict[str, Any]) -> bool: + if "degrade_on_unavailable" in params: + return bool(params.get("degrade_on_unavailable")) + return _side_effect_degrade_default() def _validate_consistency_sweep_remote_all(result: Any) -> dict[str, Any]: @@ -138,21 +232,58 @@ class StateHubContextResolver(ContextResolver): payload = { key: value for key, value in params.items() - if key not in {"required"} + if key not in {"required", "degrade_on_unavailable"} } - result = _post_json("/recently-on-scope/hourly", payload) + try: + result = _post_json("/recently-on-scope/hourly", payload) + except (httpx.HTTPError, ValueError) as exc: + if not _want_degrade(params): + raise + logger.warning( + "recently_on_scope_hourly degraded after retries: %s", exc + ) + return { + "generated": [], + "skipped": [], + "failed": [ + { + "reason": "edge_unavailable", + "detail": str(exc)[:300], + } + ], + "degraded": True, + "degraded_reason": str(exc)[:300], + } return _validate_recently_on_scope_hourly(result) if query == "consistency_sweep_remote_all": payload = { key: value for key, value in params.items() - if key not in {"required"} + if key not in {"required", "degrade_on_unavailable"} } - result = _post_json( - "/consistency/sweep/remote-all", - payload, - timeout=_SWEEP_TIMEOUT_SECONDS, - ) + try: + result = _post_json( + "/consistency/sweep/remote-all", + payload, + timeout=_SWEEP_TIMEOUT_SECONDS, + ) + except (httpx.HTTPError, ValueError) as exc: + if not _want_degrade(params): + raise + logger.warning( + "consistency_sweep_remote_all degraded after retries: %s", exc + ) + # Shape matches validator; exit_code 75 = temporary failure. + return { + "exit_code": 75, + "lock_skipped": True, + "repos_processed": [], + "skipped_clean": [], + "skipped_missing": [], + "skipped_budget": [], + "degraded": True, + "degraded_reason": str(exc)[:300], + } return _validate_consistency_sweep_remote_all(result) if query == "phase5_stabilization_check": return _phase5_stabilization_check(params) diff --git a/tests/test_state_hub_context_resolver.py b/tests/test_state_hub_context_resolver.py index 0f377f1..b200206 100644 --- a/tests/test_state_hub_context_resolver.py +++ b/tests/test_state_hub_context_resolver.py @@ -9,13 +9,25 @@ from activity_core.context_resolvers.state_hub import StateHubContextResolver class DummyResponse: - def __init__(self, payload: Any, status_error: Exception | None = None) -> None: + def __init__( + self, + payload: Any, + status_error: Exception | None = None, + status_code: int = 200, + ) -> None: self.payload = payload self.status_error = status_error + self.status_code = status_code def raise_for_status(self) -> None: if self.status_error is not None: raise self.status_error + if self.status_code >= 400: + req = httpx.Request("POST", "http://dummy") + resp = httpx.Response(self.status_code, request=req) + raise httpx.HTTPStatusError( + f"{self.status_code}", request=req, response=resp + ) def json(self) -> Any: return self.payload @@ -498,9 +510,30 @@ def test_recently_on_scope_hourly_failure_bubbles(monkeypatch) -> None: raise httpx.ConnectError("offline") monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setenv("STATE_HUB_POST_RETRIES", "1") + monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0") with pytest.raises(httpx.ConnectError): - StateHubContextResolver().resolve("recently_on_scope_hourly", None, {"range": "1h"}) + StateHubContextResolver().resolve( + "recently_on_scope_hourly", + None, + {"range": "1h", "degrade_on_unavailable": False}, + ) + + +def test_recently_on_scope_hourly_degrades(monkeypatch) -> None: + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + raise httpx.ConnectError("offline") + + monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setenv("STATE_HUB_POST_RETRIES", "2") + monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0") + + result = StateHubContextResolver().resolve( + "recently_on_scope_hourly", None, {"range": "1h"} + ) + assert result["degraded"] is True + assert result["failed"][0]["reason"] == "edge_unavailable" def test_consistency_sweep_remote_all_posts_batch(monkeypatch) -> None: @@ -544,15 +577,69 @@ def test_consistency_sweep_remote_all_failure_bubbles(monkeypatch) -> None: raise httpx.ConnectError("offline") monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setenv("STATE_HUB_POST_RETRIES", "1") + monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0") with pytest.raises(httpx.ConnectError): StateHubContextResolver().resolve( "consistency_sweep_remote_all", None, - {"max_seconds": 300}, + {"max_seconds": 300, "degrade_on_unavailable": False}, ) +def test_consistency_sweep_degrades_after_retries(monkeypatch) -> None: + calls: list[int] = [] + + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + calls.append(1) + raise httpx.ConnectError("offline") + + monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setenv("STATE_HUB_POST_RETRIES", "3") + monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0") + monkeypatch.setenv("STATE_HUB_SIDE_EFFECT_DEGRADE", "true") + + result = StateHubContextResolver().resolve( + "consistency_sweep_remote_all", + None, + {"max_seconds": 300}, + ) + assert result["degraded"] is True + assert result["exit_code"] == 75 + assert result["lock_skipped"] is True + assert len(calls) == 3 + + +def test_consistency_sweep_retries_then_succeeds(monkeypatch) -> None: + calls: list[int] = [] + + def fake_post(url: str, **kwargs: Any) -> DummyResponse: + calls.append(1) + if len(calls) < 2: + return DummyResponse({}, status_code=503) + return DummyResponse( + { + "exit_code": 0, + "lock_skipped": False, + "repos_processed": [], + }, + status_code=200, + ) + + monkeypatch.setattr(httpx, "post", fake_post) + monkeypatch.setenv("STATE_HUB_POST_RETRIES", "3") + monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0") + + result = StateHubContextResolver().resolve( + "consistency_sweep_remote_all", + None, + {"max_seconds": 300}, + ) + assert result["exit_code"] == 0 + assert len(calls) == 2 + + def test_consistency_sweep_remote_all_rejects_empty_response(monkeypatch) -> None: def fake_post(url: str, **kwargs: Any) -> DummyResponse: return DummyResponse({}) diff --git a/workplans/ACTIVITY-WP-0027-llm-connect-and-run-artifacts.md b/workplans/ACTIVITY-WP-0027-llm-connect-and-run-artifacts.md index 4049956..b588b7c 100644 --- a/workplans/ACTIVITY-WP-0027-llm-connect-and-run-artifacts.md +++ b/workplans/ACTIVITY-WP-0027-llm-connect-and-run-artifacts.md @@ -4,7 +4,7 @@ type: workplan title: "Hardened llm-connect access + run→artifact links in ops UI" domain: infotech repo: activity-core -status: active +status: finished owner: grok topic_slug: activity-core priority: high @@ -276,7 +276,7 @@ links in ops UI without hand-editing DB rows. ```task id: ACTIVITY-WP-0027-T06 -status: todo +status: done priority: medium state_hub_task_id: "52f1db5e-23db-4dcd-aaba-71c11deaf09a" ``` @@ -296,7 +296,7 @@ fixed or explicitly accepted with a follow-up id. ```task id: ACTIVITY-WP-0027-T07 -status: todo +status: done priority: low state_hub_task_id: "245955d4-83bd-4145-b71e-2e6449754e31" ``` @@ -336,7 +336,7 @@ Depends on T02 + T04 (minimum). - [x] rein-aharness on railiance01 does not depend on ad-hoc `kubectl port-forward` for llm-connect - [x] `https://activity.coulomb.social/ops/ui` shows clickable artefact links for recent FI/Binky successes (API verified; SSO UI serves same payload) - [x] Contract tests cover join + URL builder; no secrets in responses -- [ ] Remaining caveats (edge 503, Binky timers) either fixed or explicitly deferred with owner +- [x] Remaining caveats (edge 503, Binky timers) either fixed or explicitly deferred with owner ## References @@ -371,3 +371,15 @@ Port-forward deprecated for claim-loop production. ### Remaining - T06 edge-relay soft-fail for required resolvers - T07 Binky dual-clock timer retirement after clean claim days + +### T06 (2026-08-05) +- `_post_json` retries 502/503/504 + network errors (default 3×) +- Side-effect POSTs degrade after retries (`STATE_HUB_SIDE_EFFECT_DEGRADE=true`) +- Doc: `docs/edge-relay-resilience.md` + runbook pointer +- Tests: degrade + retry success paths + +### T07 (2026-08-05) +- Binky claim smoke: ops_run `eeccd98a-…` succeeded via claim loop (manual trigger) +- Prior scheduled claim success: Aug 4 `13fcbed2-…` +- Disabled user timers: `binky-rhythm-daily`, `binky-rhythm-mail`, `binky-rhythm-review` + (units remain on disk for break-glass)