fix: disambiguate TypeScript and MPEG-TS routing

openviking/parse/parser_router.py

@@ -13,7 +13,10 @@
 if TYPE_CHECKING:
     from openviking.parse.accessors.base import LocalResource
 
+from openviking.parse.accessors.base import SourceType
 from openviking.parse.base import ParseResult
+from openviking.parse.parsers.constants import CODE_EXTENSIONS
+from openviking.parse.parsers.media.constants import MEDIA_EXTENSIONS
 from openviking.parse.registry import ParserRegistry
 from openviking_cli.utils.logger import get_logger
 
@@ -65,8 +68,9 @@ async def parse(self, source: Union[str, Path, "LocalResource"], **kwargs) -> Pa
         Parse with ParserRegistry or UnderstandingAPI based on the routing decision.
         """
         source_path = self._extract_source_path(source)
+        prefer_local_code = self._is_local_code_media_conflict(source, source_path)
 
-        if self.should_use_understanding_api(source_path):
+        if self.should_use_understanding_api(source_path) and not prefer_local_code:
             display = source_path
             if isinstance(source_path, str) and source_path.startswith(("http://", "https://")):
                 display = "<url>"
@@ -83,8 +87,40 @@ async def parse(self, source: Union[str, Path, "LocalResource"], **kwargs) -> Pa
             except Exception:
                 display = "<path>"
             logger.info(f"[ParserRouter] Using internal ParserRegistry for {display}")
+
+            media_parser_name = self._explicit_remote_media_parser(source, source_path)
+            if media_parser_name:
+                parser = self._parser_registry.get_parser(media_parser_name)
+                if parser is not None:
+                    return await parser.parse(source_path, **kwargs)
+
             return await self._parser_registry.parse(source_path, **kwargs)
 
+    def _is_local_code_media_conflict(
+        self,
+        source: Union[str, Path, "LocalResource"],
+        source_path: Union[str, Path],
+    ) -> bool:
+        """Whether a local file extension should prefer code over media."""
+        if getattr(source, "source_type", None) != SourceType.LOCAL:
+            return False
+        ext = Path(source_path).suffix.lower()
+        return ext in CODE_EXTENSIONS and ext in MEDIA_EXTENSIONS
+
+    def _explicit_remote_media_parser(
+        self,
+        source: Union[str, Path, "LocalResource"],
+        source_path: Union[str, Path],
+    ) -> str | None:
+        """Restore media routing for an accessor-confirmed ambiguous HTTP file."""
+        if getattr(source, "source_type", None) != SourceType.HTTP:
+            return None
+        ext = Path(source_path).suffix.lower()
+        if ext not in CODE_EXTENSIONS or ext not in MEDIA_EXTENSIONS:
+            return None
+        url_type = getattr(source, "meta", {}).get("url_type")
+        return "video" if url_type == "download_video" else None
+
     def _extract_source_path(self, source: Union[str, Path, "LocalResource"]) -> Union[str, Path]:
         """Extract a filesystem path from the source."""
         if hasattr(source, "path"):

openviking/parse/parsers/media/utils.py

@@ -7,6 +7,7 @@
 from typing import TYPE_CHECKING, Any, Dict, Optional
 
 from openviking.core.path_variables import CalendarVariableProvider
+from openviking.parse.parsers.constants import CODE_EXTENSIONS
 from openviking.prompts import render_prompt
 from openviking.storage.viking_fs import get_viking_fs
 from openviking_cli.utils.config import get_openviking_config
@@ -68,6 +69,10 @@ def get_media_type(source_path: Optional[str], source_format: Optional[str]) ->
 
     if source_path:
         ext = Path(source_path).suffix.lower()
+        # A parser-provided non-media format is stronger evidence when a file
+        # extension is also a documented code extension (currently ``.ts``).
+        if source_format and ext in CODE_EXTENSIONS:
+            return None
         if ext in IMAGE_EXTENSIONS:
             return "image"
         elif ext in AUDIO_EXTENSIONS:
@@ -78,6 +83,29 @@ def get_media_type(source_path: Optional[str], source_format: Optional[str]) ->
     return None
 
 
+def get_resource_media_type(resource_uri: str) -> Optional[str]:
+    """Determine media type for a file already stored in OpenViking.
+
+    Ambiguous code/media extensions are media only inside the corresponding
+    generated media namespace.  This keeps TypeScript files in repositories on
+    the AST/text summary path while preserving MPEG-TS resources under
+    ``viking://resources/video/...``.
+    """
+    media_type = get_media_type(resource_uri, None)
+    if not media_type:
+        return None
+
+    ext = Path(resource_uri).suffix.lower()
+    if ext not in CODE_EXTENSIONS:
+        return media_type
+
+    media_dir = {"image": "images", "audio": "audio", "video": "video"}[media_type]
+    normalized_uri = resource_uri.replace("\\", "/")
+    if normalized_uri.startswith(f"viking://resources/{media_dir}/"):
+        return media_type
+    return None
+
+
 def get_media_base_uri(media_type: str) -> str:
     """
     Get base URI for media files.

openviking/parse/registry.py

@@ -11,6 +11,7 @@
 
 from openviking.parse.base import ParseResult
 from openviking.parse.parsers.base_parser import BaseParser
+from openviking.parse.parsers.constants import CODE_EXTENSIONS
 from openviking.parse.parsers.directory import DirectoryParser
 from openviking.parse.parsers.epub import EPubParser
 from openviking.parse.parsers.excel import ExcelParser
@@ -22,6 +23,7 @@
 from openviking.parse.parsers.legacy_doc import LegacyDocParser
 from openviking.parse.parsers.markdown import MarkdownParser
 from openviking.parse.parsers.media import AudioParser, ImageParser, VideoParser
+from openviking.parse.parsers.media.constants import MEDIA_EXTENSIONS
 from openviking.parse.parsers.pdf import PDFParser
 from openviking.parse.parsers.powerpoint import PowerPointParser
 from openviking.parse.parsers.text import TextParser
@@ -196,13 +198,27 @@ def get_parser_for_file(self, path: Union[str, Path]) -> Optional[BaseParser]:
         """
         path = Path(path)
         ext = path.suffix.lower()
+
+        # Some extensions are inherently ambiguous.  In particular, ``.ts``
+        # means TypeScript for local files but MPEG transport stream for media
+        # URLs.  Local registry calls do not carry HTTP content metadata, so
+        # prefer the documented code/text fallback here.  ParserRouter restores
+        # the media parser when an accessor explicitly classified a remote
+        # resource as media.
+        if ext in CODE_EXTENSIONS and ext in MEDIA_EXTENSIONS:
+            return None
+
         parser_name = self._extension_map.get(ext)
 
         if parser_name:
             return self._parsers.get(parser_name)
 
         return None
 
+    def get_parser(self, name: str) -> Optional[BaseParser]:
+        """Return a registered parser by name."""
+        return self._parsers.get(name)
+
     async def parse(self, source: Union[str, Path], **kwargs) -> ParseResult:
         """
         Parse a local file or content string.

openviking/parse/understanding_api.py

@@ -168,7 +168,7 @@ async def parse(self, source: Union[str, Path], instruction: str = "", **kwargs)
         result = ParseResult(
             root=root_node,
             source_path=url or source_str,
-            source_format=doc_type,
+            source_format=content_type if content_type != "text" else doc_type,
             temp_dir_path=temp_dir_path,
             parser_name="UnderstandingAPI",
             meta=task_meta,

openviking/storage/queuefs/semantic_processor.py

@@ -28,7 +28,7 @@
     generate_audio_summary,
     generate_image_summary,
     generate_video_summary,
-    get_media_type,
+    get_resource_media_type,
 )
 from openviking.prompts import render_prompt
 from openviking.server.identity import RequestContext, Role
@@ -1141,7 +1141,7 @@ async def _generate_single_file_summary(
         """
         file_name = file_path.split("/")[-1]
         llm_sem = llm_sem or asyncio.Semaphore(self.max_concurrent_llm)
-        media_type = get_media_type(file_name, None)
+        media_type = get_resource_media_type(file_path)
         if media_type == "image":
             return await generate_image_summary(file_path, file_name, llm_sem, ctx=ctx)
         elif media_type == "audio":

tests/parse/test_directory_parser_routing.py

@@ -85,6 +85,7 @@ def tmp_all_parsers(tmp_path: Path) -> Path:
                 app.py            -> text-fallback (is_text_file)
                 engine.cc         -> text-fallback
                 main.js           -> text-fallback
+                app.ts            -> text-fallback (not MPEG-TS video)
                 style.css         -> text-fallback
             config/
                 settings.yaml     -> text-fallback
@@ -115,6 +116,7 @@ def tmp_all_parsers(tmp_path: Path) -> Path:
     (tmp_path / "code" / "app.py").write_text("print(1)", encoding="utf-8")
     (tmp_path / "code" / "engine.cc").write_text("int main() { return 0; }", encoding="utf-8")
     (tmp_path / "code" / "main.js").write_text("console.log(1)", encoding="utf-8")
+    (tmp_path / "code" / "app.ts").write_text("const answer = 42", encoding="utf-8")
     (tmp_path / "code" / "style.css").write_text("body{}", encoding="utf-8")
 
     (tmp_path / "config").mkdir()
@@ -168,7 +170,16 @@ class TestParserSelection:
     # Extensions that are *processable* (via is_text_file) but have no
     # dedicated parser in the registry – they fall back to TextParser at
     # parse-time via ``ParserRegistry.parse``.
-    TEXT_FALLBACK_EXTENSIONS = {".py", ".cc", ".js", ".css", ".yaml", ".json", ".toml"}
+    TEXT_FALLBACK_EXTENSIONS = {
+        ".py",
+        ".cc",
+        ".js",
+        ".ts",
+        ".css",
+        ".yaml",
+        ".json",
+        ".toml",
+    }
 
     def test_dedicated_parsers_resolve(self, registry: ParserRegistry) -> None:
         """get_parser_for_file returns the correct class for each extension."""

tests/parse/test_media_type_routing.py

@@ -0,0 +1,17 @@
+from openviking.parse.parsers.media.utils import get_media_type, get_resource_media_type
+
+
+def test_explicit_text_parser_format_overrides_ambiguous_ts_extension():
+    assert get_media_type("example.ts", "markdown") is None
+
+
+def test_explicit_video_format_preserves_mpeg_ts():
+    assert get_media_type("example.ts", "video") == "video"
+
+
+def test_typescript_in_repository_uses_text_summary_path():
+    assert get_resource_media_type("viking://resources/org/repo/src/example.ts") is None
+
+
+def test_mpeg_ts_in_video_namespace_uses_video_summary_path():
+    assert get_resource_media_type("viking://resources/video/2026/07/19/example.ts") == "video"

tests/parse/test_parser_router.py

@@ -1,5 +1,9 @@
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
+import pytest
+
+from openviking.parse.accessors.base import LocalResource, SourceType
 from openviking.parse.parser_router import ParserRouter
 
 
@@ -17,3 +21,55 @@ def test_should_use_understanding_api_for_signed_video_url(monkeypatch):
     assert router.should_use_understanding_api(
         "https://example.com/media/video.mp4?X-Tos-Signature=abc&X-Tos-Expires=60"
     )
+
+
+@pytest.mark.asyncio
+async def test_local_typescript_bypasses_media_understanding_api(monkeypatch, tmp_path):
+    source_path = tmp_path / "example.ts"
+    source_path.write_text("const answer = 42", encoding="utf-8")
+    source = LocalResource(
+        path=source_path,
+        source_type=SourceType.LOCAL,
+        original_source=str(source_path),
+        is_temporary=False,
+    )
+    config = SimpleNamespace(parser_api=SimpleNamespace(enable=True, extensions=["ts"]))
+    monkeypatch.setattr(
+        "openviking_cli.utils.config.open_viking_config.get_openviking_config",
+        lambda: config,
+    )
+    registry = SimpleNamespace(parse=AsyncMock(return_value="text-result"))
+    router = ParserRouter(parser_registry=registry)
+    router._get_understanding_api = lambda: (_ for _ in ()).throw(
+        AssertionError("local TypeScript must not use UnderstandingAPI")
+    )
+
+    assert await router.parse(source) == "text-result"
+    registry.parse.assert_awaited_once_with(source_path)
+
+
+@pytest.mark.asyncio
+async def test_remote_mpeg_ts_uses_accessor_confirmed_video_parser(monkeypatch, tmp_path):
+    source_path = tmp_path / "download.ts"
+    source_path.write_bytes(b"mpeg-ts")
+    source = LocalResource(
+        path=source_path,
+        source_type=SourceType.HTTP,
+        original_source="https://example.com/sample.ts",
+        meta={"url_type": "download_video"},
+    )
+    config = SimpleNamespace(parser_api=SimpleNamespace(enable=False, extensions=[]))
+    monkeypatch.setattr(
+        "openviking_cli.utils.config.open_viking_config.get_openviking_config",
+        lambda: config,
+    )
+    video_parser = SimpleNamespace(parse=AsyncMock(return_value="video-result"))
+    registry = SimpleNamespace(
+        get_parser=lambda name: video_parser if name == "video" else None,
+        parse=AsyncMock(),
+    )
+    router = ParserRouter(parser_registry=registry)
+
+    assert await router.parse(source) == "video-result"
+    video_parser.parse.assert_awaited_once_with(source_path)
+    registry.parse.assert_not_awaited()

tests/storage/test_typescript_summary_routing.py

@@ -0,0 +1,40 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from openviking.storage.queuefs import semantic_processor as semantic_processor_module
+from openviking.storage.queuefs.semantic_processor import SemanticProcessor
+
+
+@pytest.mark.asyncio
+async def test_repository_typescript_uses_text_summary(monkeypatch):
+    processor = SemanticProcessor.__new__(SemanticProcessor)
+    processor.max_concurrent_llm = 1
+    processor._generate_text_summary = AsyncMock(return_value={"kind": "text"})
+    video_summary = AsyncMock(return_value={"kind": "video"})
+    monkeypatch.setattr(semantic_processor_module, "generate_video_summary", video_summary)
+
+    result = await processor._generate_single_file_summary(
+        "viking://resources/org/repo/src/example.ts"
+    )
+
+    assert result == {"kind": "text"}
+    processor._generate_text_summary.assert_awaited_once()
+    video_summary.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_mpeg_ts_resource_uses_video_summary(monkeypatch):
+    processor = SemanticProcessor.__new__(SemanticProcessor)
+    processor.max_concurrent_llm = 1
+    processor._generate_text_summary = AsyncMock(return_value={"kind": "text"})
+    video_summary = AsyncMock(return_value={"kind": "video"})
+    monkeypatch.setattr(semantic_processor_module, "generate_video_summary", video_summary)
+
+    result = await processor._generate_single_file_summary(
+        "viking://resources/video/2026/07/19/example.ts"
+    )
+
+    assert result == {"kind": "video"}
+    video_summary.assert_awaited_once()
+    processor._generate_text_summary.assert_not_awaited()