tenant.openagents/omega

1from __future__ import annotations
2
3import argparse
4from typing import Any
5
6from . import source
7
8
9def validate_build_id(build_id: str | None) -> None:
10    if build_id and source.sanitize_namespace(build_id) != build_id:
11        raise ValueError(
12            "build ids may only contain lowercase letters, numbers, '.', '_', and '-'"
13        )
14
15
16def prepare_build_request(
17    *,
18    base_sha: str | None,
19    patch_path: str | None,
20    build_id: str | None,
21    allow_untracked: bool,
22    require_clean: bool,
23    repo_url: str | None,
24    clean_source: bool = False,
25    source_label: str | None = None,
26    pre_resolved_base_sha: str | None = None,
27) -> dict[str, Any]:
28    validate_build_id(build_id)
29    source_info, patch = source.prepare_build_source(
30        base_sha=base_sha,
31        patch_path=patch_path,
32        allow_untracked=allow_untracked,
33        require_clean=require_clean,
34        repo_url=repo_url,
35        clean=clean_source,
36        source_label=source_label,
37        pre_resolved_base_sha=pre_resolved_base_sha,
38    )
39    resolved_build_id = build_id or source.compute_build_id(source_info=source_info)
40    return {
41        "build_id": resolved_build_id,
42        "base_sha": source_info["base_sha"],
43        "patch": patch,
44        "patch_sha256": source_info.get("patch_sha256"),
45        "source": source_info,
46        "toolchain": source.build_toolchain_info(),
47    }
48
49
50def resolve_source(
51    args: argparse.Namespace,
52) -> tuple[str | None, bool, str | None, str | None]:
53    """Unify the source selectors into
54    `(base_sha, clean_source, source_label, pre_resolved_base_sha)`.
55
56    Precedence: `--from` wins over `--zed-version` / `--base-sha`.
57
58      --from local         -> current HEAD + tracked patch (dev iteration)
59      --from <ref/tag/sha>  -> clean build of that ref, resolved canonically
60                               against the remote so everyone shares one build
61    """
62    repo_url = getattr(args, "repo_url", None) or source.DEFAULT_REPO_URL
63    from_source = getattr(args, "from_source", None)
64    if from_source:
65        if from_source.strip().lower() == "local":
66            return (
67                getattr(args, "base_sha", None),
68                bool(getattr(args, "clean_source", False)),
69                None,
70                None,
71            )
72        resolved = source.resolve_remote_ref(repo_url, from_source)
73        return None, True, from_source, resolved
74
75    zed_version = getattr(args, "zed_version", None)
76    if zed_version:
77        resolved = source.resolve_remote_ref(repo_url, zed_version)
78        return None, True, zed_version, resolved
79
80    return (
81        getattr(args, "base_sha", None),
82        bool(getattr(args, "clean_source", False)),
83        None,
84        None,
85    )
86