diff --git a/.gitignore b/.gitignore index 8f770c98..4a856497 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,8 @@ __pycache__/ *.db-shm *.db-wal *.db-journal +flow_generation_hold.json +flow_generation_hold.tmp .env output/ youtube/channels/ diff --git a/agent/api/flow.py b/agent/api/flow.py index 7ed3602c..18f9859b 100644 --- a/agent/api/flow.py +++ b/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(), } diff --git a/agent/config.py b/agent/config.py index 42aa2c45..574c208c 100644 --- a/agent/config.py +++ b/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")) ) diff --git a/agent/services/flow_client.py b/agent/services/flow_client.py index e16e11ab..874a2f57 100644 --- a/agent/services/flow_client.py +++ b/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, diff --git a/agent/worker/processor.py b/agent/worker/processor.py index 1acf65ba..336f6054 100644 --- a/agent/worker/processor.py +++ b/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,16 +497,20 @@ 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, @@ -509,7 +518,7 @@ async def _handle_failure(rid: str, req: dict, result: dict, retry_after: dict = # 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)) diff --git a/skills/fk-doctor.md b/skills/fk-doctor.md index a2fc34b8..8de0f611 100644 --- a/skills/fk-doctor.md +++ b/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` | diff --git a/tests/conftest.py b/tests/conftest.py index 90e1b995..922b894e 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/unit/test_flow_global_throttle.py b/tests/unit/test_flow_global_throttle.py index 9fed7740..b3498652 100644 --- a/tests/unit/test_flow_global_throttle.py +++ b/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" diff --git a/tests/unit/test_processor.py b/tests/unit/test_processor.py index c615912c..682aaa49 100644 --- a/tests/unit/test_processor.py +++ b/tests/unit/test_processor.py @@ -7,6 +7,7 @@ _is_already_completed, _mark_scene_failed, _handle_failure, + HIJACK_MAX_RETRIES, ) from agent.config import MAX_RETRIES @@ -202,6 +203,38 @@ async def test_unusual_activity_is_not_retried_as_captcha(self): assert "retry_count" not in call_kwargs[1] mock_crud.update_scene.assert_awaited_once_with("scene-001", vertical_video_status="FAILED") + @pytest.mark.asyncio + async def test_bare_unusual_activity_code_is_not_retried(self): + req = make_req(retry_count=0) + with patch("agent.worker.processor.crud") as mock_crud: + mock_crud.update_request = AsyncMock() + mock_crud.update_scene = AsyncMock() + await _handle_failure(req["id"], req, {"error": "UNUSUAL_ACTIVITY"}) + assert mock_crud.update_request.call_args[1]["status"] == "FAILED" + assert "retry_count" not in mock_crud.update_request.call_args[1] + + @pytest.mark.asyncio + async def test_hijack_is_retried_and_counted(self): + req = make_req(retry_count=0) + with patch("agent.worker.processor.crud") as mock_crud: + mock_crud.update_request = AsyncMock() + mock_crud.update_scene = AsyncMock() + await _handle_failure(req["id"], req, {"error": "[HIJACK] extension_hijack_detected"}) + kwargs = mock_crud.update_request.call_args[1] + assert kwargs["status"] == "PENDING" and kwargs["retry_count"] == 1 + mock_crud.update_scene.assert_not_awaited() + + @pytest.mark.asyncio + async def test_hijack_fails_after_the_limit(self): + req = make_req(req_type="GENERATE_IMAGE", scene_id="scene-001", + retry_count=HIJACK_MAX_RETRIES - 1) + with patch("agent.worker.processor.crud") as mock_crud: + mock_crud.update_request = AsyncMock() + mock_crud.update_scene = AsyncMock() + await _handle_failure(req["id"], req, {"error": "[HIJACK] extension_hijack_detected"}) + assert mock_crud.update_request.call_args[1]["status"] == "FAILED" + mock_crud.update_scene.assert_awaited_once_with("scene-001", vertical_image_status="FAILED") + @pytest.mark.asyncio async def test_extracts_error_message_from_nested_data(self): """Error message extraction from data.error.message should work."""