Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/guides/installing-loopx.md
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,12 @@ not necessarily active in an installed release. Archive maintainers may use
the tagged package channel and wait for the corresponding release instead of
trying to switch an installed distribution to `main`.

An install taken from a pinned full commit SHA needs no separate lineage
lookup: `loopx update check --ref <40-hex-commit>` compares that commit with the
installed manifest source commit and reports `runtime_active` when they match,
so the readback you would use to confirm your own pinned install stays
actionable.

For an installation owned by another Python package manager, `update plan`
reports that owner and its command; LoopX fails closed instead of guessing a
pip mutation. For a live source checkout, it reports the contributor installer
Expand Down
7 changes: 7 additions & 0 deletions docs/product/release-readiness.md
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,13 @@ commit with the trusted source lineage reported by `loopx doctor`:
belongs to a different `repo/ref`; the runtime-active claim must fail closed
until identity is refreshed.

A pinned full commit SHA is its own trusted target, so
`loopx update check --ref <40-hex-commit>` qualifies against the installed
manifest source commit without waiting for a branch lineage lookup. Installed
and pinned commits match: the receipt is `runtime_active`. They differ: the
receipt stays `activation_qualification_required` and names the installed
commit difference instead of the generic lineage message.

Closing a PR monitor after latest-`main` validation is valid, but the closeout
must not say the fix is active in the installed runtime unless this receipt is
`runtime_active`. Publishing a release remains a separate maintainer action.
Expand Down
47 changes: 47 additions & 0 deletions examples/loopx-update-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -427,6 +427,52 @@ def test_check_degrades_when_source_version_is_unavailable() -> None:
assert "could not be checked" in payload["recommended_action"], payload


def immutable_fresh_doctor_payload(ref: str) -> dict[str, object]:
payload = fake_fresh_doctor_payload()
freshness = payload["install_freshness"]
assert isinstance(freshness, dict)
freshness["manifest_source_ref"] = ref
freshness["freshness_source_label"] = f"example/loopx@{ref}"
freshness["freshness_source_git_commit"] = None
return payload


def test_immutable_source_ref_qualifies_activation() -> None:
pinned = INSTALLED_COMMIT
with mock.patch(
"loopx.self_update.urlopen",
return_value=FakeVersionResponse(__version__),
):
activated = build_update_plan(
repo="example/loopx",
ref=pinned,
check_only=True,
doctor_payload=immutable_fresh_doctor_payload(pinned),
)
qualification = activated["runtime_activation_qualification"]
assert qualification["decision"] == "runtime_active", activated
assert qualification["runtime_active"] is True, activated
assert qualification["target_source_commit"] == pinned, activated
assert qualification["revision_relation"] == "same", activated
assert "no update needed" in activated["recommended_action"], activated

other = TARGET_COMMIT
unproven = build_update_plan(
repo="example/loopx",
ref=other,
check_only=True,
doctor_payload=immutable_fresh_doctor_payload(other),
)
mismatch = unproven["runtime_activation_qualification"]
assert mismatch["decision"] == "activation_qualification_required", unproven
assert mismatch["target_source_commit"] == other, unproven
assert "does not match the selected immutable source commit" in mismatch["reason"], unproven
assert mismatch["successor"] == {
"required": True,
"kind": "activation_qualification",
}, unproven


def test_rollback_previous_executes_with_temp_home() -> None:
with TemporaryDirectory() as tmpdir:
home = Path(tmpdir)
Expand Down Expand Up @@ -545,6 +591,7 @@ def main() -> int:
test_fresh_check_is_noop_recommendation()
test_check_compares_selected_source_version()
test_check_degrades_when_source_version_is_unavailable()
test_immutable_source_ref_qualifies_activation()
test_rollback_previous_executes_with_temp_home()
test_rollback_restores_previous_when_doctor_fails()
test_cli_check()
Expand Down
138 changes: 138 additions & 0 deletions loopx/activation_qualification.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
"""Qualify whether the installed runtime proves a selected update source is active.

``loopx update check`` cannot call a release active just because a newer tag or
branch exists: the installed archive has to carry source lineage that proves the
selected update source reached the runtime. This module owns that decision so the
question keeps one home instead of growing inside the self-update command surface.
"""

from __future__ import annotations

import re
from typing import Any


RUNTIME_ACTIVATION_QUALIFICATION_SCHEMA_VERSION = (
"loopx_runtime_activation_qualification_v0"
)
_FULL_COMMIT_PATTERN = re.compile(r"[0-9a-fA-F]{40}")


def _immutable_source_commit(source: dict[str, Any]) -> str | None:
"""Return the selected source commit when the ref is an immutable full SHA.

A full commit SHA is its own commit identity: ``scripts/install.sh`` installs
it without a branch-resolution service, so the qualification can compare it
against the installed manifest instead of reporting that the trusted target
source lineage is unavailable.
"""

ref = str(source.get("ref") or "")
return ref.lower() if _FULL_COMMIT_PATTERN.fullmatch(ref) else None


def runtime_activation_qualification(

Check failure on line 34 in loopx/activation_qualification.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 19 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=huangruiteng_loopx&issues=AaCl7QIX8nySzipwn0jm&open=AaCl7QIX8nySzipwn0jm&pullRequest=4460
*,
install_freshness: dict[str, Any],
source: dict[str, Any],
) -> dict[str, Any]:
"""Return the typed activation qualification for an archive-snapshot install."""

installed_commit = install_freshness.get("manifest_source_git_commit")
target_commit = install_freshness.get("freshness_source_git_commit")
revision_relation = install_freshness.get("manifest_source_freshness_relation")
selected_commit = _immutable_source_commit(source)
if selected_commit and not target_commit:
# An immutable ref names its own commit, so the qualification resolves it
# the same way the installer does instead of reporting that the
# installed-versus-target lineage is unavailable.
target_commit = selected_commit
if (
isinstance(installed_commit, str)
and installed_commit.lower() == selected_commit
):
revision_relation = "same"
qualified_repo = install_freshness.get("manifest_source_repo")
qualified_ref = install_freshness.get("manifest_source_ref")
selected_repo = source.get("repo")
selected_ref = source.get("ref")
package_matches_runtime = install_freshness.get(
"manifest_package_version_matches_runtime"
)
requires_upgrade = install_freshness.get("requires_upgrade")
has_commit_pair = all(
isinstance(commit, str) and bool(commit)
for commit in (installed_commit, target_commit)
)
source_identity_matches = all(
isinstance(value, str) and bool(value)
for value in (qualified_repo, qualified_ref, selected_repo, selected_ref)
) and (
str(qualified_repo).removesuffix(".git").lower()
== str(selected_repo).removesuffix(".git").lower()
and str(qualified_ref).removeprefix("refs/heads/")
== str(selected_ref).removeprefix("refs/heads/")
)

if package_matches_runtime is False:
decision = "release_or_install_successor_required"
runtime_active: bool | None = False
successor_kind = "release_or_install"
reason = "release manifest package version does not match the active runtime"
elif not source_identity_matches:
decision = "activation_qualification_required"
runtime_active = None
successor_kind = "activation_qualification"
reason = "trusted source lineage does not identify the selected update source"
elif has_commit_pair and (
installed_commit == target_commit or revision_relation == "installed_ahead"
):
decision = "runtime_active"
runtime_active = True
successor_kind = None
reason = "installed source contains the trusted target source commit"
elif has_commit_pair and revision_relation in {"installed_behind", "diverged"}:
decision = "release_or_install_successor_required"
runtime_active = False
successor_kind = "release_or_install"
reason = "installed source does not contain the trusted target source commit"
elif has_commit_pair and selected_commit:
decision = "activation_qualification_required"
runtime_active = None
successor_kind = "activation_qualification"
reason = (
"installed source commit does not match the selected immutable source commit"
)
else:
decision = "activation_qualification_required"
runtime_active = None
successor_kind = "activation_qualification"
reason = "trusted installed-versus-target source lineage is unavailable"

return {
"schema_version": RUNTIME_ACTIVATION_QUALIFICATION_SCHEMA_VERSION,
"decision": decision,
"runtime_active": runtime_active,
"installed_release_id": install_freshness.get("release_id"),
"installed_version": install_freshness.get("current_version"),
"installed_source_commit": installed_commit,
"target_source_label": install_freshness.get("freshness_source_label"),
"target_source_commit": target_commit,
"revision_relation": revision_relation,
"qualified_source": {
"repo": qualified_repo,
"ref": qualified_ref,
},
"source_identity_matches": source_identity_matches,
"package_version_matches_runtime": package_matches_runtime,
"requires_upgrade": requires_upgrade,
"selected_source": {
"repo": selected_repo,
"ref": selected_ref,
},
"successor": {
"required": runtime_active is not True,
"kind": successor_kind,
},
"reason": reason,
}
96 changes: 5 additions & 91 deletions loopx/self_update.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@
from urllib.parse import quote
from urllib.request import Request, urlopen

from .activation_qualification import (
RUNTIME_ACTIVATION_QUALIFICATION_SCHEMA_VERSION,
runtime_activation_qualification,
)
from .doctor import collect_doctor
from .install_contract import NO_CLONE_INSTALL_URL
from .runtime_activation import restart_services_for_runtime_activation
Expand All @@ -22,9 +26,6 @@
DEFAULT_UPDATE_REF = "stable"
ROLLBACK_PREVIOUS_ALIAS = "previous"
SOURCE_VERSION_CHECK_SCHEMA_VERSION = "loopx_source_version_check_v0"
RUNTIME_ACTIVATION_QUALIFICATION_SCHEMA_VERSION = (
"loopx_runtime_activation_qualification_v0"
)
SOURCE_VERSION_CHECK_TIMEOUT_SECONDS = 3
SOURCE_VERSION_READ_LIMIT_BYTES = 64 * 1024
PERSISTED_PYTHON_FILENAME = ".loopx-python"
Expand Down Expand Up @@ -214,93 +215,6 @@ def _source_version_check(source: dict[str, Any]) -> dict[str, Any]:
}


def _runtime_activation_qualification(
*,
install_freshness: dict[str, Any],
source: dict[str, Any],
) -> dict[str, Any]:
installed_commit = install_freshness.get("manifest_source_git_commit")
target_commit = install_freshness.get("freshness_source_git_commit")
revision_relation = install_freshness.get("manifest_source_freshness_relation")
qualified_repo = install_freshness.get("manifest_source_repo")
qualified_ref = install_freshness.get("manifest_source_ref")
selected_repo = source.get("repo")
selected_ref = source.get("ref")
package_matches_runtime = install_freshness.get(
"manifest_package_version_matches_runtime"
)
requires_upgrade = install_freshness.get("requires_upgrade")
has_commit_pair = all(
isinstance(commit, str) and bool(commit)
for commit in (installed_commit, target_commit)
)
source_identity_matches = all(
isinstance(value, str) and bool(value)
for value in (qualified_repo, qualified_ref, selected_repo, selected_ref)
) and (
str(qualified_repo).removesuffix(".git").lower()
== str(selected_repo).removesuffix(".git").lower()
and str(qualified_ref).removeprefix("refs/heads/")
== str(selected_ref).removeprefix("refs/heads/")
)

if package_matches_runtime is False:
decision = "release_or_install_successor_required"
runtime_active: bool | None = False
successor_kind = "release_or_install"
reason = "release manifest package version does not match the active runtime"
elif not source_identity_matches:
decision = "activation_qualification_required"
runtime_active = None
successor_kind = "activation_qualification"
reason = "trusted source lineage does not identify the selected update source"
elif has_commit_pair and (
installed_commit == target_commit or revision_relation == "installed_ahead"
):
decision = "runtime_active"
runtime_active = True
successor_kind = None
reason = "installed source contains the trusted target source commit"
elif has_commit_pair and revision_relation in {"installed_behind", "diverged"}:
decision = "release_or_install_successor_required"
runtime_active = False
successor_kind = "release_or_install"
reason = "installed source does not contain the trusted target source commit"
else:
decision = "activation_qualification_required"
runtime_active = None
successor_kind = "activation_qualification"
reason = "trusted installed-versus-target source lineage is unavailable"

return {
"schema_version": RUNTIME_ACTIVATION_QUALIFICATION_SCHEMA_VERSION,
"decision": decision,
"runtime_active": runtime_active,
"installed_release_id": install_freshness.get("release_id"),
"installed_version": install_freshness.get("current_version"),
"installed_source_commit": installed_commit,
"target_source_label": install_freshness.get("freshness_source_label"),
"target_source_commit": target_commit,
"revision_relation": revision_relation,
"qualified_source": {
"repo": qualified_repo,
"ref": qualified_ref,
},
"source_identity_matches": source_identity_matches,
"package_version_matches_runtime": package_matches_runtime,
"requires_upgrade": requires_upgrade,
"selected_source": {
"repo": selected_repo,
"ref": selected_ref,
},
"successor": {
"required": runtime_active is not True,
"kind": successor_kind,
},
"reason": reason,
}


def _release_root_from_doctor(doctor_payload: dict[str, Any]) -> str | None:
package = (
doctor_payload.get("package")
Expand Down Expand Up @@ -651,7 +565,7 @@ def build_update_plan(
else None
)
runtime_activation = (
_runtime_activation_qualification(
runtime_activation_qualification(
install_freshness=install_freshness,
source=source,
)
Expand Down
2 changes: 1 addition & 1 deletion loopx/semantics/inventory_v0.json
Original file line number Diff line number Diff line change
Expand Up @@ -902,7 +902,7 @@
]
},
"summary": {
"source_files": 1176,
"source_files": 1177,
"python_enums": 103,
"python_closed_sets": 495,
"python_literal_aliases": 8,
Expand Down
Loading