fix(repo): PATH@REV takes a revision with a slash; flush progress per task

dokimi/changelog.d/20261021_120000_napetrov_scan-rev-unbuffered.md

@@ -0,0 +1,4 @@
+### Fixed
+
+- **`PATH@REV` takes a revision with a slash** — `repo scan .@origin/main` (and `refs/heads/x`) read the whole spec as a path and failed with "not a directory"; a branch with `/` is now a revision for every `repo` command that takes a source. A suffix that is no revision (`a..b`, `x//y`, trailing `/`) still stays in the path, and an existing local directory with an `@` in its path (`ws/job@2/repo`) is read whole.
+- **`repo tasks check` and `repo run` progress is flushed per task** — under a pipe (CI) the per-task lines were block-buffered and all appeared at exit unless `PYTHONUNBUFFERED` was set.

dokimi/dokimi/commands/repo_run.py

@@ -142,6 +142,12 @@ def _pair(text: str) -> tuple[str, str]:
     return name, path
 
 
+def _progress(line: str) -> None:
+    """One progress line, flushed: under a pipe (CI) stdout is block-buffered and the per-task
+    lines would otherwise all appear when the command exits."""
+    print(line, flush=True)
+
+
 def _errors():
     from dokimi.repos import ManifestError, SourceError
     from dokimi.repos.container import ContainerError
@@ -162,7 +168,7 @@ def cmd_repo_tasks_check(args: argparse.Namespace) -> None:
         tasks = load_tasks(ctx.content_dir, ctx.manifest, args.task)
         print(f"{len(tasks)} task(s), image {ctx.image}{' (warm: with network)' if args.warm else ''}",
               file=sys.stderr)
-        results = run_check(ctx, tasks, out / "check", warm=args.warm, jobs=args.jobs)
+        results = run_check(ctx, tasks, out / "check", warm=args.warm, jobs=args.jobs, log=_progress)
     except _errors() as exc:
         print(f"✗ {exc}", file=sys.stderr)
         sys.exit(1)
@@ -223,7 +229,7 @@ def _run_locked(args: argparse.Namespace, arms: list[str], batch_dir: Path, lock
         print(f"{profile.name} {version} in {ctx.image}; {len(tasks)} task(s) x {len(arms)} arm(s) x "
               f"{len(options)} model(s) x {args.reps} rep(s) -> {batch_dir}", file=sys.stderr)
         summary = run_batch(ctx, tasks, arms, options, args.reps, batch_dir, jobs=args.jobs, force=args.force,
-                            max_batch_usd=args.max_batch_usd)
+                            max_batch_usd=args.max_batch_usd, log=_progress)
     except _errors() as exc:
         print(f"✗ {exc}", file=sys.stderr)
         sys.exit(1)

dokimi/dokimi/repos/source.py

@@ -25,10 +25,12 @@
 
 from dokimi import process
 
-# A revision is a SHA, tag or single-segment branch, optionally with ^ / ~N
-# suffixes. No "/" (so "user@host:path" URLs are not split) and no leading "-"
-# (so a revision can never be read as a git option).
-_REV_RE = re.compile(r"^[A-Za-z0-9._][A-Za-z0-9._^~-]*$")
+# A revision is a SHA, tag or branch (a "/" is allowed: origin/main, refs/heads/x), optionally
+# with ^ / ~N suffixes. No leading "-" (so a revision can never be read as a git option);
+# _is_rev also refuses "..", "//" and a trailing "/". A "user@host" URL is kept whole by
+# _has_location_path, which splits at the last "@" only when what is left is still a location;
+# in a remote spec a "/" in the revision needs a ref-like start (_remote_split_ok).
+_REV_RE = re.compile(r"^[A-Za-z0-9._][A-Za-z0-9._/^~-]*$")  # "/" for origin/main, refs/heads/x
 _SHA_RE = re.compile(r"^[0-9a-f]{7,40}$")
 _REMOTE_PREFIXES = ("https://", "ssh://", "file://", "git@")
 _GIT_TIMEOUT_S = 1800
@@ -54,7 +56,10 @@ def parse(cls, spec: str) -> RepoSource:
             raise SourceError(f"invalid repository source: {spec!r}")
         location, rev = spec, None
         head, sep, tail = spec.rpartition("@")
-        if sep and head and _REV_RE.match(tail):
+        # An existing directory is read whole: jenkins ws/job@2/repo, /home/a@corp.com/proj.
+        if (sep and head and _is_rev(tail) and _has_location_path(head)
+                and (_remote_split_ok(spec, head, tail) if spec.startswith(_REMOTE_PREFIXES)
+                     else not Path(spec).is_dir())):
             location, rev = head, tail
         remote = location.startswith(_REMOTE_PREFIXES)
         if location.startswith("http://"):
@@ -430,9 +435,37 @@ def _fetch_refs(git_dir: Path) -> None:
          "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
 
 
+def _has_location_path(head: str) -> bool:
+    """False for the head of an unpinned remote URL cut at its ``user@``: ``ssh://git`` in
+    ``ssh://git@github.com/o/r.git`` has no path, so the tail is a host and not a revision."""
+    for prefix in ("https://", "ssh://", "file://"):
+        if head.startswith(prefix):
+            return "/" in head[len(prefix):]
+    return True
+
+
+_REMOTE_REV_ROOTS = ("origin", "upstream", "refs", "remotes", "heads", "tags")
+
+
+def _remote_split_ok(spec: str, head: str, tail: str) -> bool:
+    """Whether an ``@`` in a remote spec starts the revision. A remote path may hold an ``@``
+    (``file:///tmp/a@b/c``, ``git@host/o/r``), so a ``/`` in the tail makes it a revision only
+    when it starts like a ref (``origin/main``, ``refs/heads/x``); ``git@host`` needs an ``@``
+    left in the head."""
+    if spec.startswith("git@") and "@" not in head:
+        return False
+    return "/" not in tail or tail.split("/", 1)[0] in _REMOTE_REV_ROOTS
+
+
+def _is_rev(rev: str) -> bool:
+    """A revision as a source spec or ``resolve_rev`` takes it: a SHA, tag, branch (``origin/main``,
+    ``refs/heads/x``) or ``HEAD~1``. No ``..`` (a range), ``//`` or trailing ``/``."""
+    return bool(_REV_RE.match(rev)) and ".." not in rev and "//" not in rev and not rev.endswith("/")
+
+
 def resolve_rev(git_dir: Path, rev: str) -> str:
     """Full commit SHA for ``rev``; raises :class:`SourceError` if unknown."""
-    if not _REV_RE.match(rev):
+    if not _is_rev(rev):
         raise SourceError(f"invalid revision: {rev!r}")
     return _git(git_dir, "rev-parse", "--verify", "--quiet", f"{rev}^{{commit}}").strip()
 

dokimi/tests/test_repo_manifest.py

@@ -643,3 +643,91 @@ def test_a_mistyped_manifest_path_is_an_error_not_a_default_scan(tmp_path, capsy
         args.func(args)
     assert exc.value.code == 1
     assert "manifest not found" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("rev", ["origin/main", "refs/heads/x", "HEAD~1", "v1.2", "a492e703^"])
+def test_source_parse_takes_a_rev_with_a_slash(tmp_path, rev):
+    src = RepoSource.parse(f"{tmp_path}@{rev}")
+    assert (src.location, src.rev, src.remote) == (str(tmp_path), rev, False)
+    remote = RepoSource.parse(f"git@github.com:o/r.git@{rev}")
+    assert (remote.location, remote.rev) == ("git@github.com:o/r.git", rev)
+
+
+@pytest.mark.parametrize("rev", ["a..b", "origin//main", "origin/", "-x"])
+def test_source_parse_leaves_a_non_rev_suffix_in_the_path(tmp_path, rev):
+    (tmp_path / f"d@{rev}").mkdir(parents=True)
+    src = RepoSource.parse(f"{tmp_path}/d@{rev}")
+    assert src.rev is None and src.location == f"{tmp_path}/d@{rev}"
+
+
+@pytest.mark.parametrize("url", [
+    "ssh://git@github.com/o/r.git", "https://user@github.com/o/r.git",
+    "file:///home/a@corp.com/proj", "file:///tmp/a@b/c", "https://github.com/o@x/r",
+    "git@github.com/o/r", "file://a@main", "https://h/o/r@/abs"])
+def test_an_unpinned_remote_url_with_a_user_is_not_split_at_its_at_sign(url):
+    with pytest.raises(SourceError, match="must be pinned"):
+        RepoSource.parse(url)
+
+
+@pytest.mark.parametrize("url", ["ssh://git@github.com/o/r.git", "https://user@github.com/o/r.git"])
+def test_a_pinned_remote_url_with_a_user_splits_at_the_last_at_sign(url):
+    src = RepoSource.parse(f"{url}@origin/main")
+    assert (src.location, src.rev, src.remote) == (url, "origin/main", True)
+
+
+@pytest.mark.parametrize("spec, location, rev", [
+    ("ssh://git@host/o/r@origin/main", "ssh://git@host/o/r", "origin/main"),
+    ("git@host:o/r@sha1234", "git@host:o/r", "sha1234"),
+    ("git@host:o/r@refs/heads/x", "git@host:o/r", "refs/heads/x"),
+    ("ssh://git@host/o/r@upstream/main", "ssh://git@host/o/r", "upstream/main"),
+    ("ssh://git@host/o/r@remotes/origin/main", "ssh://git@host/o/r", "remotes/origin/main"),
+    ("ssh://git@host/o/r@heads/x", "ssh://git@host/o/r", "heads/x"),
+    ("ssh://git@host/o/r@tags/v1", "ssh://git@host/o/r", "tags/v1"),
+    ("file:///home/a@corp.com/proj@abc1234", "file:///home/a@corp.com/proj", "abc1234"),
+    ("file:///tmp/a@b/c@v1", "file:///tmp/a@b/c", "v1"),
+])
+def test_a_pinned_remote_spec_with_an_at_sign_in_its_path_splits_at_the_last_one(spec, location, rev):
+    src = RepoSource.parse(spec)
+    assert (src.location, src.rev, src.remote) == (location, rev, True)
+
+
+@pytest.mark.parametrize("spec", ["git@host:o/r@sha1234", "file:/tmp/x@main"])
+def test_a_remote_spec_is_split_even_if_a_directory_of_that_name_exists(tmp_path, monkeypatch, spec):
+    (tmp_path / spec).mkdir(parents=True)
+    monkeypatch.chdir(tmp_path)
+    assert RepoSource.parse(spec if spec.startswith("git@") else "file:///tmp/x@main").rev
+
+
+def test_a_local_rev_does_not_start_with_a_slash(tmp_path):
+    (tmp_path / "d").mkdir()
+    with pytest.raises(SourceError, match="not a directory"):
+        RepoSource.parse(f"{tmp_path}/d@/x")
+
+
+@pytest.mark.parametrize("rev", ["a..b", "x//y", "y/", "-x"])
+def test_resolve_rev_refuses_what_is_no_revision(tmp_path, rev):
+    from dokimi.repos.source import resolve_rev
+
+    root = _repo(tmp_path)
+    _git_commit(root)
+    with pytest.raises(SourceError, match="invalid revision"):
+        resolve_rev(root, rev)
+
+
+def test_resolve_rev_takes_a_branch_with_a_slash(tmp_path):
+    from dokimi.repos.source import resolve_rev
+
+    root = _repo(tmp_path)
+    sha = _git_commit(root)
+    subprocess.run(["git", "-C", str(root), "branch", "origin/main"], check=True)
+    assert resolve_rev(root, "origin/main") == sha
+
+
+@pytest.mark.parametrize("sub", ["ws/job@2/repo", "home/alice@corp.com/proj", "gomod/r@v1.2.3/skills",
+                                 "node_modules/@types/node", "d@HEAD"])
+def test_an_existing_directory_with_an_at_sign_is_read_whole(tmp_path, sub):
+    (tmp_path / sub).mkdir(parents=True)
+    src = RepoSource.parse(f"{tmp_path}/{sub}")
+    assert (src.location, src.rev) == (f"{tmp_path}/{sub}", None)
+    pinned = RepoSource.parse(f"{tmp_path}/{sub}@origin/main")
+    assert (pinned.location, pinned.rev) == (f"{tmp_path}/{sub}", "origin/main")

dokimi/tests/test_repo_progress.py

@@ -0,0 +1,37 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Progress lines of `repo tasks check` / `repo run` are flushed one by one, not at exit."""
+
+import select
+import subprocess
+import sys
+
+
+def test_progress_line_is_flushed_under_a_pipe():
+    code = ("import sys, time\n"
+            "from dokimi.commands.repo_run import _progress\n"
+            "_progress('t1: OK')\n"
+            "time.sleep(30)\n")
+    proc = subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True,
+                            env={"PATH": "", "PYTHONPATH": ":".join(sys.path)})
+    try:
+        assert proc.stdout is not None
+        # A flushed line arrives at once; a buffered one stays in the child until it exits (30 s).
+        ready, _, _ = select.select([proc.stdout], [], [], 5)
+        assert ready, "no progress line within 5 s: the line was not flushed"
+        assert proc.stdout.readline() == "t1: OK\n"
+    finally:
+        proc.kill()
+        proc.wait()
+        if proc.stdout is not None:
+            proc.stdout.close()
+
+
+def test_commands_hand_the_flushing_logger_to_the_runners():
+    import inspect
+
+    from dokimi.commands import repo_run
+
+    src = inspect.getsource(repo_run)
+    assert src.count("log=_progress") == 2