feat(release): InspectedResult.recommended_action centralizes exclusion decision

Add a derived 'recommended_action' property on InspectedResult that
collapses the orchestrator's go / wait / skip decision into one value:

- 'skip'      → no main_video, or media_type == 'other'
- 'ask_user'  → media_type == 'unknown', or road == 'path_of_pain'
- 'process'   → confident parse with a main video on disk

The ordering is part of the contract (skip > ask_user > process) —
documented in the property docstring.

Until now every consumer (workflows, the agent, the orchestrator
sketch) had to re-derive this from the road / media_type / main_video
triple, with subtle drift between sites. One place, one rule.

Exposed through the analyze_release tool so the LLM can route on it.
Spec YAML updated to describe the new field.

Suite: 1083 passed (+6 new tests in tests/application/test_inspect.py
covering the four branches and the precedence rules).
This commit is contained in:
2026-05-21 07:54:17 +02:00
parent b7979c0f8b
commit 5107cb32c0
5 changed files with 152 additions and 2 deletions
+11
View File
@@ -48,6 +48,17 @@ callers).
### Added ### Added
- **`InspectedResult.recommended_action` property** — derived hint that
collapses the orchestrator's go / wait / skip decision into a single
value (``"process"`` / ``"ask_user"`` / ``"skip"``). Centralizes the
exclusion logic that was previously dispersed across road /
media_type / main_video checks at each call site. Ordering is part of
the contract: ``skip`` (no main video, or media_type == ``"other"``)
wins over ``ask_user`` (media_type == ``"unknown"`` or road ==
``"path_of_pain"``) which wins over ``process``. Surfaced through the
``analyze_release`` tool so the LLM can route on it directly.
6 new tests in ``tests/application/test_inspect.py`` cover the four
branches and the precedence rules.
- **`LanguageRepository` port** in `alfred.domain.shared.ports`. Structural - **`LanguageRepository` port** in `alfred.domain.shared.ports`. Structural
Protocol covering `from_iso`, `from_any`, `all`, `__contains__`, `__len__` Protocol covering `from_iso`, `from_any`, `all`, `__contains__`, `__len__`
— the surface previously coupled to the concrete `LanguageRegistry`. — the surface previously coupled to the concrete `LanguageRegistry`.
+1
View File
@@ -238,6 +238,7 @@ def analyze_release(release_name: str, source_path: str) -> dict[str, Any]:
"probe_used": result.probe_used, "probe_used": result.probe_used,
"confidence": result.report.confidence, "confidence": result.report.confidence,
"road": result.report.road, "road": result.report.road,
"recommended_action": result.recommended_action,
} }
@@ -82,3 +82,4 @@ returns:
probe_used: True when ffprobe successfully enriched the result. probe_used: True when ffprobe successfully enriched the result.
confidence: Parser confidence score, 0100 (higher = more reliable). confidence: Parser confidence score, 0100 (higher = more reliable).
road: "Parser road: 'easy' (group schema matched), 'shitty' (heuristic but acceptable), or 'path_of_pain' (low confidence — ask the user before auto-routing)." road: "Parser road: 'easy' (group schema matched), 'shitty' (heuristic but acceptable), or 'path_of_pain' (low confidence — ask the user before auto-routing)."
recommended_action: "Orchestrator hint: 'process' (go straight to resolve_*_destination), 'ask_user' (media_type unknown or road=path_of_pain — confirm with the user first), or 'skip' (no main video, or media_type=other — nothing to organize)."
+48 -2
View File
@@ -62,6 +62,21 @@ from alfred.domain.shared.media import MediaInfo
from alfred.domain.shared.ports import MediaProber from alfred.domain.shared.ports import MediaProber
# Media types for which a probe carries no useful information.
_NON_PROBABLE_MEDIA_TYPES = frozenset({"unknown", "other"})
# Media types for which there's nothing for the organizer to do.
# ``other`` covers things like games / ISOs / archives sitting on the
# downloads folder. ``unknown`` does NOT belong here — those need a
# user decision, not a skip.
_SKIPPABLE_MEDIA_TYPES = frozenset({"other"})
# Roads that signal the parser couldn't reach a confident answer on its
# own. ``Road`` values are kept as strings on the report to avoid a
# cross-package import here.
_ASK_USER_ROADS = frozenset({"path_of_pain"})
@dataclass(frozen=True) @dataclass(frozen=True)
class InspectedResult: class InspectedResult:
"""The full picture of a release: parsed name + filesystem reality. """The full picture of a release: parsed name + filesystem reality.
@@ -85,6 +100,10 @@ class InspectedResult:
- ``probe_used`` — ``True`` iff ``media_info`` is non-``None`` and - ``probe_used`` — ``True`` iff ``media_info`` is non-``None`` and
``enrich_from_probe`` actually ran. Explicit flag so callers ``enrich_from_probe`` actually ran. Explicit flag so callers
don't have to re-derive the condition. don't have to re-derive the condition.
- ``recommended_action`` — derived hint for the orchestrator (see
property docstring). Encodes the exclusion / clarification /
go-ahead decision in one place so downstream callers don't
re-implement the same checks.
""" """
parsed: ParsedRelease parsed: ParsedRelease
@@ -94,9 +113,36 @@ class InspectedResult:
media_info: MediaInfo | None media_info: MediaInfo | None
probe_used: bool probe_used: bool
@property
def recommended_action(self) -> str:
"""Return one of ``"skip"`` / ``"ask_user"`` / ``"process"``.
# Media types for which a probe carries no useful information. - ``"skip"`` — nothing to organize:
_NON_PROBABLE_MEDIA_TYPES = frozenset({"unknown", "other"}) * the source has no main video file, **or**
* ``media_type`` is ``"other"`` (games / ISOs / archives).
- ``"ask_user"`` — a decision is required before any action:
* ``media_type`` is ``"unknown"`` (parser couldn't classify), **or**
* the parse landed on ``Road.PATH_OF_PAIN``
(low-confidence, malformed name, etc.).
- ``"process"`` — everything else: a confident parse with a
usable media type and a main video on disk. The orchestrator
can move straight to the planning step.
The check ordering matters: ``"skip"`` wins over ``"ask_user"``
because if there's no video to organize, no question to the
user can change that. ``"ask_user"`` then wins over
``"process"`` because a confident parse alone isn't enough if
the type or road still flag uncertainty.
"""
if self.main_video is None:
return "skip"
if self.parsed.media_type.value in _SKIPPABLE_MEDIA_TYPES:
return "skip"
if self.parsed.media_type.value == "unknown":
return "ask_user"
if self.report.road in _ASK_USER_ROADS:
return "ask_user"
return "process"
def inspect_release( def inspect_release(
+91
View File
@@ -263,3 +263,94 @@ class TestFrozen:
pass pass
else: # pragma: no cover else: # pragma: no cover
raise AssertionError("InspectedResult should be frozen") raise AssertionError("InspectedResult should be frozen")
# --------------------------------------------------------------------------- #
# recommended_action #
# --------------------------------------------------------------------------- #
class TestRecommendedAction:
"""``recommended_action`` collapses the orchestrator's go / wait /
skip decision into a single property. The check ordering is part
of the contract (skip wins over ask_user, ask_user wins over
process) — see the property docstring."""
def test_skip_when_no_main_video(self, tmp_path: Path) -> None:
# Folder with no video at all → main_video is None → skip.
folder = tmp_path / _MOVIE_NAME
folder.mkdir()
(folder / "readme.txt").write_text("hi")
result = inspect_release(_MOVIE_NAME, folder, _KB, _RaisingProber())
assert result.main_video is None
assert result.recommended_action == "skip"
def test_skip_when_media_type_other(self, tmp_path: Path) -> None:
# Folder with only non-video files (ISO) → media_type == "other"
# AND main_video is None (find_main_video filters by video ext).
# Both branches resolve to "skip"; this asserts the contract holds.
folder = tmp_path / _MOVIE_NAME
folder.mkdir()
(folder / "disc.iso").write_bytes(b"")
result = inspect_release(_MOVIE_NAME, folder, _KB, _RaisingProber())
assert result.parsed.media_type == "other"
assert result.recommended_action == "skip"
def test_ask_user_when_media_type_unknown(self, tmp_path: Path) -> None:
# Mixed video + non-video → detect_media_type returns "unknown".
folder = tmp_path / _MOVIE_NAME
folder.mkdir()
(folder / "movie.mkv").write_bytes(b"")
(folder / "extras.iso").write_bytes(b"")
result = inspect_release(
_MOVIE_NAME, folder, _KB, _StubProber(_media_info_1080p_h264())
)
assert result.parsed.media_type == "unknown"
assert result.recommended_action == "ask_user"
def test_ask_user_when_path_of_pain_road(self, tmp_path: Path) -> None:
# Malformed name (forbidden chars) → road == "path_of_pain".
name = "garbage@#%name"
folder = tmp_path / "release"
folder.mkdir()
(folder / "movie.mkv").write_bytes(b"")
result = inspect_release(
name, folder, _KB, _StubProber(_media_info_1080p_h264())
)
assert result.report.road == "path_of_pain"
# main_video is found but the road still flags uncertainty.
assert result.main_video is not None
assert result.recommended_action == "ask_user"
def test_process_for_confident_movie(self, tmp_path: Path) -> None:
folder = tmp_path / _MOVIE_NAME
folder.mkdir()
(folder / "movie.mkv").write_bytes(b"")
result = inspect_release(
_MOVIE_NAME, folder, _KB, _StubProber(_media_info_1080p_h264())
)
assert result.parsed.media_type == "movie"
assert result.report.road in ("easy", "shitty")
assert result.recommended_action == "process"
def test_process_for_confident_tv_show(self, tmp_path: Path) -> None:
folder = tmp_path / _TV_NAME
folder.mkdir()
(folder / "episode.mkv").write_bytes(b"")
result = inspect_release(
_TV_NAME, folder, _KB, _StubProber(_media_info_1080p_h264())
)
assert result.parsed.media_type == "tv_show"
assert result.recommended_action == "process"