fix(flow): keep the unusual-activity hold across restarts, bound hijack retries

.gitignore

@@ -4,6 +4,8 @@ __pycache__/
 *.db-shm
 *.db-wal
 *.db-journal
+flow_generation_hold.json
+flow_generation_hold.tmp
 .env
 output/
 youtube/channels/

agent/api/flow.py

@@ -201,19 +201,15 @@ async def extension_status():
 
 @router.post("/clear-hijack")
 async def clear_hijack_cooldown():
-    """Reset the generation cooldown triggered by extension_hijack_detected.
+    """Lift the generation hold (hijack or unusual-activity) by hand, including
+    the copy on disk.
 
-    Call this after deploying the bypass fix to immediately resume generation
-    without waiting for the cooldown to expire.
+    Only do this once the account generates normally in the Flow UI again:
+    submitting into an active unusual-activity block may prolong it.
     """
-    import time as _time
-    client = get_flow_client()
-    old_until = client._generation_unusual_until
-    client._generation_unusual_until = 0.0
-    was_active = old_until > 0.0 and old_until > _time.monotonic()
     return {
         "cleared": True,
-        "was_active": was_active,
+        "was_active": get_flow_client().clear_generation_hold(),
     }
 
 

agent/config.py

@@ -41,9 +41,13 @@
 FLOW_GENERATION_MAX_CONCURRENT = max(
     1, int(os.environ.get("FLOW_GENERATION_MAX_CONCURRENT", "1"))
 )
+# After Google answers PUBLIC_ERROR_UNUSUAL_ACTIVITY, no generation is submitted
+# for this long. The hold is written to FLOW_GENERATION_HOLD_PATH so a restart
+# does not lift it (#67): retrying into the block may be what prolongs it.
 FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S = max(
-    0.0, float(os.environ.get("FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S", "120"))
+    0.0, float(os.environ.get("FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S", "3600"))
 )
+FLOW_GENERATION_HOLD_PATH = BASE_DIR / "flow_generation_hold.json"
 FLOW_SESSION_PROJECT_IDLE_S = max(
     300.0, float(os.environ.get("FLOW_SESSION_PROJECT_IDLE_S", "7200"))
 )

agent/services/flow_client.py

@@ -29,7 +29,7 @@
     FLOW_PROJECT_ID, FLOW_ALLOW_DEGRADED,
     DEFAULT_PAYGATE_TIER,
     FLOW_GENERATION_MIN_INTERVAL_S, FLOW_GENERATION_MAX_CONCURRENT,
-    FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S,
+    FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S, FLOW_GENERATION_HOLD_PATH,
 )
 from agent import config as _config
 from agent.services import flow_batch as fb
@@ -72,6 +72,8 @@ def __init__(self):
         self._generation_unusual_until = 0.0
         self._generation_last_unusual_at: Optional[float] = None
         self._generation_last_unusual_rpc: Optional[str] = None
+        self._generation_hold_reason: Optional[str] = None
+        self._load_generation_hold()
         # WS stats
         self._ws_connect_count = 0
         self._ws_disconnect_count = 0
@@ -204,12 +206,66 @@ def set_flow_key(self, key: str):
     def connected(self) -> bool:
         return bool(self._extensions)
 
+    def _set_generation_hold(self, seconds: float, reason: str, rpcid: str) -> None:
+        """Stop generation submits for ``seconds`` and record it on disk, so a
+        restart keeps the hold. A longer hold already in place is kept, with
+        its reason."""
+        until = time.monotonic() + seconds
+        if until <= self._generation_unusual_until:
+            return
+        self._generation_unusual_until = until
+        self._generation_last_unusual_at = time.time()
+        self._generation_last_unusual_rpc = rpcid
+        self._generation_hold_reason = reason
+        tmp = FLOW_GENERATION_HOLD_PATH.with_suffix(".tmp")
+        try:
+            tmp.write_text(json.dumps({
+                "until": time.time() + seconds,
+                "reason": reason,
+                "rpc": rpcid,
+                "at": self._generation_last_unusual_at,
+            }))
+            tmp.replace(FLOW_GENERATION_HOLD_PATH)
+        except OSError as exc:
+            logger.warning("Could not persist the generation hold: %s", exc)
+
+    def _load_generation_hold(self) -> None:
+        try:
+            hold = json.loads(FLOW_GENERATION_HOLD_PATH.read_text())
+            remaining = float(hold["until"]) - time.time()
+        except FileNotFoundError:
+            return
+        except (OSError, ValueError, KeyError, TypeError) as exc:
+            logger.warning("Ignoring unreadable generation hold %s: %s",
+                           FLOW_GENERATION_HOLD_PATH, exc)
+            return
+        if remaining <= 0:
+            return
+        self._generation_unusual_until = time.monotonic() + remaining
+        self._generation_last_unusual_at = hold.get("at")
+        self._generation_last_unusual_rpc = hold.get("rpc")
+        self._generation_hold_reason = hold.get("reason")
+        logger.warning("Generation hold restored from disk (%s): %.0fs left",
+                       self._generation_hold_reason, remaining)
+
+    def clear_generation_hold(self) -> bool:
+        """Lift the hold by hand. Returns whether one was active."""
+        was_active = self._generation_unusual_until > time.monotonic()
+        self._generation_unusual_until = 0.0
+        self._generation_hold_reason = None
+        try:
+            FLOW_GENERATION_HOLD_PATH.unlink(missing_ok=True)
+        except OSError as exc:
+            logger.warning("Could not remove the generation hold file: %s", exc)
+        return was_active
+
     @property
     def generation_guard_status(self) -> dict:
         remaining = max(0.0, self._generation_unusual_until - time.monotonic())
         return {
             "cooldown_active": remaining > 0,
             "cooldown_remaining_s": round(remaining, 3),
+            "cooldown_reason": self._generation_hold_reason if remaining > 0 else None,
             "last_unusual_activity_at": self._generation_last_unusual_at,
             "last_unusual_activity_rpc": self._generation_last_unusual_rpc,
         }
@@ -567,13 +623,9 @@ async def batch_rpc(self, rpcid: str, freq: str,
                 # x2a trap: the token was poisoned with action
                 # "extension_hijack_detected". This is a system-level trap,
                 # NOT an account issue — don't apply the long cooldown.
-                # Short pause lets the bypass re-initialise on next tab.
-                self._generation_unusual_until = max(
-                    self._generation_unusual_until,
-                    time.monotonic() + 30.0,
-                )
-                self._generation_last_unusual_at = time.time()
-                self._generation_last_unusual_rpc = rpcid
+                # Short pause lets the bypass re-initialise on next tab; the
+                # worker gives up on the request after a few of these.
+                self._set_generation_hold(30.0, "extension_hijack_detected", rpcid)
                 logger.error(
                     "[HIJACK] extension_hijack_detected — captcha bypass may "
                     "have failed; pausing generation submits for 30s"
@@ -584,12 +636,9 @@ async def batch_rpc(self, rpcid: str, freq: str,
                 else:
                     result["error"] = "[HIJACK] extension_hijack_detected"
             elif is_unusual:
-                self._generation_unusual_until = max(
-                    self._generation_unusual_until,
-                    time.monotonic() + FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S,
+                self._set_generation_hold(
+                    FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S, "unusual_activity", rpcid,
                 )
-                self._generation_last_unusual_at = time.time()
-                self._generation_last_unusual_rpc = rpcid
                 logger.warning(
                     "Google unusual-activity block detected; pausing generation submits for %.0fs",
                     FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S,

agent/worker/processor.py

@@ -20,6 +20,11 @@
 
 logger = logging.getLogger(__name__)
 
+# A failed captcha bypass gives up once the request's retry_count reaches this,
+# instead of retrying forever (#67). The count is shared with other retries,
+# which only makes it stop sooner.
+HIJACK_MAX_RETRIES = 3
+
 _API_CALL_TYPES = {"GENERATE_IMAGE", "REGENERATE_IMAGE", "EDIT_IMAGE",
                    "GENERATE_VIDEO", "REGENERATE_VIDEO", "GENERATE_VIDEO_REFS", "UPSCALE_VIDEO",
                    "GENERATE_CHARACTER_IMAGE", "REGENERATE_CHARACTER_IMAGE",
@@ -492,24 +497,28 @@ async def _handle_failure(rid: str, req: dict, result: dict, retry_after: dict =
     error_lower = str(error_msg).lower()
 
     # [HIJACK] extension_hijack_detected — the captcha bypass failed and the
-    # token was poisoned by x2a. This is a system-level trap, NOT an account
-    # issue. Don't burn retry count — the bypass may succeed on the next
-    # attempt once the tab re-initializes.
+    # token was poisoned by x2a. The bypass may succeed once the tab
+    # re-initializes, but each retry is another submit Google can see, so
+    # give up after HIJACK_MAX_RETRIES (#67).
     if "[hijack]" in error_lower or "extension_hijack" in error_lower:
-        await crud.update_request(rid, status="PENDING", error_message=str(error_msg))
-        logger.error(
-            "Request %s [HIJACK] captcha bypass failed — will retry without "
-            "counting (system error, not account error): %s",
-            rid[:8], error_msg,
-        )
+        retry = req.get("retry_count", 0) + 1
+        if retry < HIJACK_MAX_RETRIES:
+            await crud.update_request(rid, status="PENDING", retry_count=retry, error_message=str(error_msg))
+            logger.error("Request %s [HIJACK] captcha bypass failed (retry %d/%d): %s",
+                         rid[:8], retry, HIJACK_MAX_RETRIES, error_msg)
+        else:
+            await crud.update_request(rid, status="FAILED", error_message=str(error_msg))
+            await _mark_scene_failed(req)
+            logger.error("Request %s FAILED after %d [HIJACK] attempts: %s",
+                         rid[:8], retry, error_msg)
         return
 
     # PUBLIC_ERROR_UNUSUAL_ACTIVITY is a Google anti-abuse/session trust block,
     # not an ordinary CAPTCHA mint failure. Retrying it in the generic CAPTCHA
     # loop only creates more generation submits while Google is asking us to
     # slow down, so stop this request and require an explicit resubmit after
     # the session/network has recovered.
-    if "public_error_unusual_activity" in error_lower or "unusual activity" in error_lower:
+    if "unusual_activity" in error_lower or "unusual activity" in error_lower:
         await crud.update_request(rid, status="FAILED", error_message=str(error_msg))
         await _mark_scene_failed(req)
         logger.error("Request %s FAILED (Google unusual-activity block; manual recovery required): %s",
@@ -531,7 +540,7 @@ async def _handle_failure(rid: str, req: dict, result: dict, retry_after: dict =
         return
 
     # reCAPTCHA errors: retry up to 10 times — deferred dict in main loop handles delay
-    if "captcha" in error_lower or "recaptcha" in error_lower or "unusual_activity" in error_lower:
+    if "captcha" in error_lower or "recaptcha" in error_lower:
         retry = req.get("retry_count", 0) + 1
         if retry < 10:
             await crud.update_request(rid, status="PENDING", retry_count=retry, error_message=str(error_msg))

skills/fk-doctor.md

@@ -150,7 +150,8 @@ Detection lives in `agent/worker/_parsing.py:_is_error`. A result is treated as
 | `NO_INJECTION_RESULT` | `chrome.scripting.executeScript` resolved with no frame result (`background.js:540`), so the envelope never ran — the Flow tab went away or was still booting at the moment of the call. Most likely on the first captcha-bearing RPC after an agent restart | Retried with backoff, counts against `MAX_RETRIES` | Usually transient — retry first. Rule out the lookalikes before digging: a failed mint reports `CAPTCHA_FAILED`, an oversized payload is not it (`MAX_RPC_TEXT` is 32 MB), and a structured-clone failure surfaces as 500, not 502. If it repeats, pin the Flow tab and reload the extension |
 | `NO_FLOW_PROJECT` | A low-level caller reached batchexecute without a project id | **Terminal — not retried** | Use `POST /api/projects` to create a real Flow project, or call a public `/api/flow/*` endpoint without `project_id` so the session-project lease resolves one automatically |
 | `UNSUPPORTED_ON_BATCH_API` | A capability whose payload was never captured off the new UI, all on the Veo path: **video upscale**, **Veo r2v**, **Veo start+end-frame chaining**. Every Omni mode is ported, so this never names Omni any more | **Terminal — not retried** | For chaining and r2v, `FLOW_ALLOW_DEGRADED=1` falls back to plain i2v off the start frame. Upscale has no fallback. Real fix: capture the payload — `docs/CAPTURE.md` |
-| `PUBLIC_ERROR_UNUSUAL_ACTIVITY` | Google rejected the generation as unusual activity / reCAPTCHA risk evaluation | **Do not auto-retry** — repeated submits can prolong the block | Pause generation submits, keep the browser/session/network stable, and retry manually only after the trust signal has had time to recover |
+| `PUBLIC_ERROR_UNUSUAL_ACTIVITY` | Google rejected the generation as unusual activity / reCAPTCHA risk evaluation | **Do not auto-retry** — repeated submits can prolong the block | FlowKit holds all generation submits for `FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S` (default 3600s); the hold is saved to `flow_generation_hold.json` and survives a restart — see `GET /api/flow/status` → `generation_throttle`. Keep the browser/session/network stable. Lift it early (`POST /api/flow/clear-hijack`) only once manual generation in the Flow UI works again |
+| `[HIJACK] extension_hijack_detected` | The page's reCAPTCHA wrapper marked the token as coming from an extension; the bypass did not capture the pristine `execute` | Retried after a 30s hold, **at most 3 attempts**, then `FAILED` | Reload the Flow tab and resubmit once. If it keeps failing, stop and check that manual generation in the Flow UI still works before trying again |
 | `no ogiZ0b envelope in response` | The RPC answered but not with the payload we came for — usually a signed-out page returning an HTML redirect | Retried with backoff | Re-sign in on the Flow tab |
 | `Polling timeout after Ns: Media not found.` | The job never produced media inside the budget | Terminal after `MAX_RETRIES` | The quoted complaint is a **diagnostic, not the cause** — finished jobs report it too. Check the Flow UI: if the clip is there, raise `VIDEO_POLL_TIMEOUT` |
 

tests/conftest.py

@@ -164,3 +164,11 @@ async def test_db(tmp_path, monkeypatch):
     yield tmp_path / "test.db"
     await schema.close_db()
     schema._db_connection = None
+
+
+@pytest.fixture(autouse=True)
+def _generation_hold_path(tmp_path, monkeypatch):
+    """Keep FlowClient's persisted generation hold out of the real agent dir."""
+    path = tmp_path / "flow_generation_hold.json"
+    monkeypatch.setattr("agent.services.flow_client.FLOW_GENERATION_HOLD_PATH", path)
+    return path

tests/unit/test_flow_global_throttle.py

@@ -74,3 +74,81 @@ async def fake_send(method, params, timeout=300):
     result = await client.batch_rpc("meta", "x")
     assert result["status"] == 200
     assert calls == 1
+
+
+async def _unusual(client, monkeypatch):
+    async def fake_send(method, params, timeout=300):
+        return {"status": 200, "data": "PUBLIC_ERROR_UNUSUAL_ACTIVITY"}
+
+    monkeypatch.setattr(client, "_send", fake_send)
+    await client.batch_rpc("a", "x", captcha_action=fb.CAPTCHA_IMAGE)
+
+
+@pytest.mark.asyncio
+async def test_unusual_activity_hold_survives_a_restart(monkeypatch, _generation_hold_path):
+    monkeypatch.setattr(fc, "FLOW_GENERATION_MIN_INTERVAL_S", 0.0)
+    monkeypatch.setattr(fc, "FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S", 3600.0)
+    await _unusual(fc.FlowClient(), monkeypatch)
+    assert _generation_hold_path.exists()
+
+    restarted = fc.FlowClient()
+    calls = 0
+
+    async def fake_send(method, params, timeout=300):
+        nonlocal calls
+        calls += 1
+        return {"status": 200, "data": "ok"}
+
+    monkeypatch.setattr(restarted, "_send", fake_send)
+    result = await restarted.batch_rpc("b", "y", captcha_action=fb.CAPTCHA_VIDEO)
+    assert result["status"] == 429 and calls == 0
+    status = restarted.generation_guard_status
+    assert status["cooldown_reason"] == "unusual_activity"
+    assert 3590 < status["cooldown_remaining_s"] <= 3600
+
+
+def test_expired_or_unreadable_hold_is_ignored(_generation_hold_path):
+    _generation_hold_path.write_text('{"until": 1, "reason": "unusual_activity"}')
+    assert fc.FlowClient().generation_guard_status["cooldown_active"] is False
+    _generation_hold_path.write_text("not json")
+    assert fc.FlowClient().generation_guard_status["cooldown_active"] is False
+
+
+@pytest.mark.asyncio
+async def test_clear_lifts_the_hold_and_its_file(monkeypatch, _generation_hold_path):
+    monkeypatch.setattr(fc, "FLOW_GENERATION_MIN_INTERVAL_S", 0.0)
+    client = fc.FlowClient()
+    await _unusual(client, monkeypatch)
+    assert client.clear_generation_hold() is True
+    assert not _generation_hold_path.exists()
+    assert fc.FlowClient().generation_guard_status["cooldown_active"] is False
+    assert client.clear_generation_hold() is False
+
+
+@pytest.mark.asyncio
+async def test_hijack_sets_a_short_hold(monkeypatch, _generation_hold_path):
+    monkeypatch.setattr(fc, "FLOW_GENERATION_MIN_INTERVAL_S", 0.0)
+    client = fc.FlowClient()
+
+    async def fake_send(method, params, timeout=300):
+        return {"status": 200, "error": "extension_hijack_detected"}
+
+    monkeypatch.setattr(client, "_send", fake_send)
+    result = await client.batch_rpc("a", "x", captcha_action=fb.CAPTCHA_IMAGE)
+    assert result["error"].startswith("[HIJACK]")
+    status = client.generation_guard_status
+    assert status["cooldown_reason"] == "extension_hijack_detected"
+    assert 0 < status["cooldown_remaining_s"] <= 30
+
+
+@pytest.mark.asyncio
+async def test_a_shorter_hold_keeps_the_longer_one_and_its_reason(monkeypatch, _generation_hold_path):
+    monkeypatch.setattr(fc, "FLOW_GENERATION_MIN_INTERVAL_S", 0.0)
+    monkeypatch.setattr(fc, "FLOW_UNUSUAL_ACTIVITY_COOLDOWN_S", 3600.0)
+    client = fc.FlowClient()
+    await _unusual(client, monkeypatch)
+    client._set_generation_hold(30.0, "extension_hijack_detected", "b")
+    status = client.generation_guard_status
+    assert status["cooldown_reason"] == "unusual_activity"
+    assert status["cooldown_remaining_s"] > 3590
+    assert fc.FlowClient().generation_guard_status["cooldown_reason"] == "unusual_activity"