test(26): cover Phase 26 patch coverage gaps (94.92% → 96.36%)

Codecov flagged 68 lines missing in the Phase 26 patch. Adds 19 targeted tests
across 8 files; per-file patch coverage rises sharply:

- agent_worker.py: 68.29% → 97.56% (whoami retry exhaustion, role mismatch,
  shutdown cleanup, module-import RuntimeError on missing PHAZE_AGENT_QUEUE)
- agent_client.py: 86.41% → 97.53% (upsert_files / put_metadata /
  put_fingerprint happy paths)
- execution.py:    89.74% → 100%   (4 best-effort log/PATCH failure paths)
- controller.py:   78.12% → 100%   (shutdown disposes engine + closes client)
- agent_tracklists.py: 88.24% → 100% (409 concurrent-writer poll exhaustion)
- functions.py:    88.46% → 100%   (malformed mood/style prediction skips)
- agent_files.py:  78.57% → 94.12% (non-music skip + enqueue failure swallow)

Two log-content assertions replaced with mock-call assertions because caplog
record propagation is fragile when other tests in the suite reconfigure root
logger handlers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: SimplicityGuy

tests/test_routers/test_agent_files.py

@@ -214,3 +214,40 @@ async def test_same_chunk_duplicate_paths_dedup(authenticated_client: AsyncClien
     assert response.status_code == 200, response.text
     result = await session.execute(select(sa_func.count()).select_from(FileRecord))
     assert result.scalar_one() == 1
+
+
+@pytest.mark.asyncio
+async def test_no_enqueue_for_non_music_file_type(smoke_app_and_router: tuple[AsyncClient, AsyncMock], seed_test_agent: tuple[Agent, str]) -> None:
+    """Non-music/video file types (e.g., .txt, .jpg) must skip the enqueue path even on INSERT."""
+    client, mock_router = smoke_app_and_router
+    chunk = {
+        "files": [
+            _make_record(path="/test/docs/readme.txt", ext="txt"),
+            _make_record(path="/test/docs/cover.jpg", ext="jpg"),
+        ],
+    }
+    response = await client.post("/api/internal/agent/files", json=chunk)
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["inserted"] == 2
+    assert body["enqueued"] == 0
+    mock_router.enqueue_for_agent.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_enqueue_exception_does_not_abort_response(
+    smoke_app_and_router: tuple[AsyncClient, AsyncMock], seed_test_agent: tuple[Agent, str]
+) -> None:
+    """Enqueue failure must be swallowed + counted as `enqueued=0` -- DB commit already succeeded."""
+    client, mock_router = smoke_app_and_router
+    mock_router.enqueue_for_agent.side_effect = RuntimeError("redis is sick")
+
+    chunk = {"files": [_make_record(path="/test/music/a.mp3")]}
+    response = await client.post("/api/internal/agent/files", json=chunk)
+
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["inserted"] == 1
+    assert body["enqueued"] == 0
+    # Handler attempted the enqueue (got to the side_effect) before catching + continuing.
+    mock_router.enqueue_for_agent.assert_awaited_once()

tests/test_routers/test_agent_tracklists.py

@@ -277,3 +277,46 @@ async def test_tracklist_unknown_token_returns_403(
             },
         )
     assert r.status_code == 403
+
+
+@pytest.mark.integration
+async def test_tracklist_concurrent_writer_returns_409_after_poll_exhaustion(
+    session: AsyncSession,
+    seed_test_agent: tuple[Agent, str],
+    redis_client: redis_async.Redis,
+) -> None:
+    """Concurrent-writer path: req_key already locked + resp_key never populated -> 409 after poll budget."""
+    from phaze.routers.agent_tracklists import _REQ_PREFIX, _TTL_SECONDS
+
+    _agent, raw_token = seed_test_agent
+    file_id = await _seed_file(session, _agent.id)
+    request_id = uuid.uuid4()
+
+    # Simulate a concurrent writer that has acquired the lock but never written the response.
+    await redis_client.set(f"{_REQ_PREFIX}{request_id}", "1", nx=True, ex=_TTL_SECONDS)
+
+    # Reduce the poll budget so the test does not actually wait 500ms.
+    import phaze.routers.agent_tracklists as router_mod
+
+    original_max = router_mod._CONCURRENT_POLL_MAX_ATTEMPTS
+    original_interval = router_mod._CONCURRENT_POLL_INTERVAL_S
+    router_mod._CONCURRENT_POLL_MAX_ATTEMPTS = 2
+    router_mod._CONCURRENT_POLL_INTERVAL_S = 0.001
+    try:
+        async with _make_client(session, redis_client, raw_token) as ac:
+            r = await ac.post(
+                "/api/internal/agent/tracklists",
+                json={
+                    "file_id": str(file_id),
+                    "source": "fingerprint",
+                    "external_id": f"fp-concurrent-{file_id.hex[:8]}",
+                    "request_id": str(request_id),
+                    "tracks": [{"position": 1}],
+                },
+            )
+    finally:
+        router_mod._CONCURRENT_POLL_MAX_ATTEMPTS = original_max
+        router_mod._CONCURRENT_POLL_INTERVAL_S = original_interval
+
+    assert r.status_code == 409, r.text
+    assert "duplicate in-flight request" in r.text

tests/test_services/test_agent_client.py

@@ -192,3 +192,59 @@ async def test_whoami_returns_agent_identity_model(client):  # type: ignore[no-u
     assert identity.name == "File Server 01"
     assert identity.scan_roots == ["/data/music"]
     assert route.call_count == 1
+
+
+@respx.mock
+async def test_upsert_files_posts_chunk_and_parses_response(client):  # type: ignore[no-untyped-def]
+    from phaze.schemas.agent_files import FileUpsertChunk, FileUpsertRecord, FileUpsertResponse
+
+    record = FileUpsertRecord(
+        sha256_hash="0" * 64,
+        original_path="/m/a.mp3",
+        original_filename="a.mp3",
+        current_path="/m/a.mp3",
+        file_type="mp3",
+        file_size=1000,
+    )
+    chunk = FileUpsertChunk(files=[record])
+    route = respx.post(f"{_BASE_URL}/api/internal/agent/files").mock(
+        return_value=httpx.Response(200, json={"agent_id": "a1", "upserted": 1, "inserted": 1, "enqueued": 1}),
+    )
+    resp = await client.upsert_files(chunk)
+    assert isinstance(resp, FileUpsertResponse)
+    assert resp.agent_id == "a1"
+    assert resp.upserted == 1
+    assert resp.inserted == 1
+    assert resp.enqueued == 1
+    assert route.call_count == 1
+
+
+@respx.mock
+async def test_put_metadata_uses_path_id_and_parses_response(client):  # type: ignore[no-untyped-def]
+    from phaze.schemas.agent_metadata import MetadataWriteRequest, MetadataWriteResponse
+
+    file_id = uuid.uuid4()
+    route = respx.put(f"{_BASE_URL}/api/internal/agent/metadata/{file_id}").mock(
+        return_value=httpx.Response(200, json={"agent_id": "a1", "file_id": str(file_id)}),
+    )
+    resp = await client.put_metadata(file_id, MetadataWriteRequest(artist="X", title="Y"))
+    assert isinstance(resp, MetadataWriteResponse)
+    assert resp.agent_id == "a1"
+    assert resp.file_id == file_id
+    assert route.call_count == 1
+
+
+@respx.mock
+async def test_put_fingerprint_includes_engine_in_url_and_parses_response(client):  # type: ignore[no-untyped-def]
+    from phaze.schemas.agent_fingerprint import FingerprintWriteRequest, FingerprintWriteResponse
+
+    file_id = uuid.uuid4()
+    engine = "audfprint"
+    route = respx.put(f"{_BASE_URL}/api/internal/agent/fingerprints/{file_id}/{engine}").mock(
+        return_value=httpx.Response(200, json={"agent_id": "a1", "file_id": str(file_id), "engine": engine}),
+    )
+    resp = await client.put_fingerprint(file_id, engine, FingerprintWriteRequest(status="success"))
+    assert isinstance(resp, FingerprintWriteResponse)
+    assert resp.agent_id == "a1"
+    assert resp.engine == engine
+    assert route.call_count == 1

tests/test_task_split.py

@@ -57,3 +57,36 @@ def test_agent_worker_does_not_import_phaze_database() -> None:
         check=False,
     )
     assert result.returncode == 0, f"agent_worker import contaminated sys.modules:\nstdout={result.stdout}\nstderr={result.stderr}"
+
+
+def test_agent_worker_module_import_fails_when_phaze_agent_queue_unset() -> None:
+    """Module-import-time guard: missing PHAZE_AGENT_QUEUE raises RuntimeError before SAQ event loop starts.
+
+    Runs in a subprocess because the module-level Queue construction is one-shot
+    and would otherwise be cached for the whole pytest session via sys.modules.
+    """
+    script = textwrap.dedent("""
+        import os
+        import sys
+        os.environ["PHAZE_ROLE"] = "agent"
+        os.environ["PHAZE_AGENT_API_URL"] = "http://localhost:8000"
+        os.environ["PHAZE_AGENT_TOKEN"] = "phaze_agent_test-token-1234567890abcdef"
+        os.environ["PHAZE_AGENT_SCAN_ROOTS"] = "/tmp"
+        os.environ["PHAZE_REDIS_URL"] = "redis://localhost:6379/0"
+        os.environ.pop("PHAZE_AGENT_QUEUE", None)
+        try:
+            import phaze.tasks.agent_worker  # noqa: F401
+        except RuntimeError as exc:
+            sys.stdout.write(str(exc))
+            sys.exit(0)
+        sys.exit(1)
+    """)
+    result = subprocess.run(  # noqa: S603  # trusted input
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        timeout=20,
+        check=False,
+    )
+    assert result.returncode == 0, f"expected RuntimeError at import; got rc={result.returncode}\nstdout={result.stdout}\nstderr={result.stderr}"
+    assert "PHAZE_AGENT_QUEUE" in result.stdout

tests/test_tasks/test_agent_startup_banner.py

@@ -112,3 +112,71 @@ async def test_agent_worker_startup_raises_on_queue_token_mismatch(
     ctx: dict[str, Any] = {}
     with pytest.raises(RuntimeError, match="queue/token mismatch"):
         await aw.startup(ctx)
+
+
+@pytest.mark.asyncio
+async def test_whoami_with_retry_raises_runtime_error_after_exhaustion(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """_whoami_with_retry: every attempt raises AgentApiError -> RuntimeError after budget exhausted."""
+    from phaze.services.agent_client import AgentApiError
+    import phaze.tasks.agent_worker as aw
+
+    # Shrink the retry budget to keep the test fast (~0s sleep total).
+    monkeypatch.setattr(aw, "_WHOAMI_BACKOFF_S", (0.0, 0.0))
+
+    fake_client = AsyncMock()
+    fake_client.whoami = AsyncMock(side_effect=AgentApiError("simulated down"))
+
+    with pytest.raises(RuntimeError, match="exhausted retry budget"):
+        await aw._whoami_with_retry(fake_client)
+
+    # 2 backoff attempts + 1 final = 3 calls
+    assert fake_client.whoami.await_count == 3
+
+
+@pytest.mark.asyncio
+async def test_startup_raises_when_role_is_not_agent(monkeypatch: pytest.MonkeyPatch) -> None:
+    """startup() must raise RuntimeError when get_settings() returns ControlSettings."""
+    import phaze.tasks.agent_worker as aw
+
+    # Return a non-AgentSettings instance so the isinstance() check trips.
+    monkeypatch.setattr(aw, "get_settings", lambda: MagicMock(name="ControlSettings"))
+
+    ctx: dict[str, Any] = {}
+    with pytest.raises(RuntimeError, match="agent_worker requires PHAZE_ROLE=agent"):
+        await aw.startup(ctx)
+
+
+@pytest.mark.asyncio
+async def test_shutdown_closes_pool_engines_and_client() -> None:
+    """shutdown() must shutdown the process pool, close each orchestrator engine, and close the api_client."""
+    import phaze.tasks.agent_worker as aw
+
+    pool = MagicMock()
+    engine_a = MagicMock()
+    engine_a.close = AsyncMock()
+    engine_b_no_close = MagicMock(spec=[])  # no .close attr -- exercise hasattr() False branch
+    orchestrator = MagicMock(engines=[engine_a, engine_b_no_close])
+    api_client = AsyncMock()
+    api_client.close = AsyncMock()
+
+    ctx: dict[str, Any] = {
+        "process_pool": pool,
+        "fingerprint_orchestrator": orchestrator,
+        "api_client": api_client,
+    }
+    await aw.shutdown(ctx)
+
+    pool.shutdown.assert_called_once_with(wait=True)
+    engine_a.close.assert_awaited_once()
+    api_client.close.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_shutdown_tolerates_missing_ctx_keys() -> None:
+    """shutdown() must no-op when ctx is empty (none of the keys were set during startup)."""
+    import phaze.tasks.agent_worker as aw
+
+    # Should not raise.
+    await aw.shutdown({})

tests/test_tasks/test_controller_startup_banner.py

@@ -47,3 +47,30 @@ async def test_controller_startup_logs_role_banner(
     assert "queue=controller" in text, f"banner missing queue=controller: {text!r}"
     # Verify the W4 fix landed: ctx["queue"] is stashed
     assert "queue" in ctx, "controller.startup did not stash ctx['queue'] (W4)"
+
+
+@pytest.mark.asyncio
+async def test_controller_shutdown_disposes_engine_and_closes_discogs_client() -> None:
+    """shutdown() must dispose task_engine and close discogs_client when present in ctx."""
+    from unittest.mock import AsyncMock
+
+    from phaze.tasks import controller
+
+    engine = MagicMock()
+    engine.dispose = AsyncMock()
+    discogs_client = MagicMock()
+    discogs_client.close = AsyncMock()
+
+    ctx: dict[str, Any] = {"task_engine": engine, "discogs_client": discogs_client}
+    await controller.shutdown(ctx)
+
+    engine.dispose.assert_awaited_once()
+    discogs_client.close.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_controller_shutdown_tolerates_missing_ctx_keys() -> None:
+    """shutdown() must no-op when startup never ran (empty ctx)."""
+    from phaze.tasks import controller
+
+    await controller.shutdown({})

tests/test_tasks/test_execute_approved_batch.py

@@ -229,3 +229,107 @@ async def test_execute_approved_batch_requires_scan_roots(tmp_path: Path, monkey
     with pytest.raises(RuntimeError, match="scan_roots"):
         await execute_approved_batch({"api_client": api}, **payload.model_dump(mode="json"))
     api.patch_proposal_state.assert_not_awaited()
+
+
+async def test_execute_approved_batch_tolerates_post_execution_log_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Best-effort: POST execution-log failure does NOT abort the file op (lines 105-108)."""
+    _patch_settings(monkeypatch, [str(tmp_path)])
+    api = _make_api_client_mock()
+    api.post_execution_log = AsyncMock(side_effect=RuntimeError("audit log down"))
+
+    orig_paths, proposed_paths = _seed_files(tmp_path, 1)
+    proposals = [
+        ExecuteBatchProposalItem(
+            proposal_id=uuid.uuid4(),
+            file_id=uuid.uuid4(),
+            original_path=str(orig_paths[0]),
+            proposed_path=str(proposed_paths[0]),
+        ),
+    ]
+    payload = ExecuteApprovedBatchPayload(batch_id=uuid.uuid4(), agent_id="a", proposals=proposals)
+    result = await execute_approved_batch({"api_client": api}, **payload.model_dump(mode="json"))
+
+    # File op still ran and proposal still marked executed
+    assert result["status"] == "completed"
+    assert proposed_paths[0].exists()
+    assert not orig_paths[0].exists()
+    assert api.patch_proposal_state.await_args.args[1].proposal_state == "executed"
+
+
+async def test_execute_approved_batch_tolerates_patch_completed_log_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Best-effort: PATCH completed log failure does NOT prevent SUCCESS report (lines 140-141)."""
+    _patch_settings(monkeypatch, [str(tmp_path)])
+    api = _make_api_client_mock()
+    api.patch_execution_log = AsyncMock(side_effect=RuntimeError("patch died"))
+
+    orig_paths, proposed_paths = _seed_files(tmp_path, 1)
+    proposals = [
+        ExecuteBatchProposalItem(
+            proposal_id=uuid.uuid4(),
+            file_id=uuid.uuid4(),
+            original_path=str(orig_paths[0]),
+            proposed_path=str(proposed_paths[0]),
+        ),
+    ]
+    payload = ExecuteApprovedBatchPayload(batch_id=uuid.uuid4(), agent_id="a", proposals=proposals)
+    result = await execute_approved_batch({"api_client": api}, **payload.model_dump(mode="json"))
+
+    assert result["status"] == "completed"
+    # patch_proposal_state still called with executed state
+    assert api.patch_proposal_state.await_args.args[1].proposal_state == "executed"
+
+
+async def test_execute_approved_batch_tolerates_patch_failed_log_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Best-effort: PATCH failed log failure does NOT prevent FAILURE report (lines 173-174)."""
+    _patch_settings(monkeypatch, [str(tmp_path)])
+    api = _make_api_client_mock()
+
+    # First PATCH (sets in_progress is via POST, second PATCH after file-op-failure goes to status=failed)
+    # Make patch_execution_log raise on EVERY call so the "failed log" branch raises.
+    api.patch_execution_log = AsyncMock(side_effect=RuntimeError("patch died"))
+
+    # Force file-op failure via missing source.
+    missing = tmp_path / "missing.mp3"
+    proposed = tmp_path / "new" / "missing.mp3"
+    proposals = [
+        ExecuteBatchProposalItem(
+            proposal_id=uuid.uuid4(),
+            file_id=uuid.uuid4(),
+            original_path=str(missing),
+            proposed_path=str(proposed),
+        ),
+    ]
+    payload = ExecuteApprovedBatchPayload(batch_id=uuid.uuid4(), agent_id="a", proposals=proposals)
+    result = await execute_approved_batch({"api_client": api}, **payload.model_dump(mode="json"))
+
+    assert result["error_count"] == 1
+    # Failure still reported via patch_proposal_state(failed)
+    assert api.patch_proposal_state.await_args.args[1].proposal_state == "failed"
+
+
+async def test_execute_approved_batch_tolerates_failure_report_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Last-line defense: even if patch_proposal_state(failed) raises, the batch returns cleanly (lines 189-192)."""
+    _patch_settings(monkeypatch, [str(tmp_path)])
+    api = _make_api_client_mock()
+    api.patch_proposal_state = AsyncMock(side_effect=RuntimeError("state-machine API down"))
+
+    # Force file-op failure -- this exercises the failed-PATCH-of-failure path.
+    missing = tmp_path / "missing.mp3"
+    proposed = tmp_path / "new" / "missing.mp3"
+    proposals = [
+        ExecuteBatchProposalItem(
+            proposal_id=uuid.uuid4(),
+            file_id=uuid.uuid4(),
+            original_path=str(missing),
+            proposed_path=str(proposed),
+        ),
+    ]
+    payload = ExecuteApprovedBatchPayload(batch_id=uuid.uuid4(), agent_id="a", proposals=proposals)
+
+    # Should NOT raise -- the inner try/except wraps the failure report.
+    result = await execute_approved_batch({"api_client": api}, **payload.model_dump(mode="json"))
+
+    assert result["processed_count"] == 1
+    assert result["error_count"] == 1
+    # Handler reached patch_proposal_state (the side_effect fired) before swallowing.
+    api.patch_proposal_state.assert_awaited_once()