Compare commits

...
Author SHA1 Message Date
sysadminandClaude Opus 4.8 61c3a57df5 fix(mcp): consume canonical target root in cross-repository operations (Closes #741)
#706 introduced the immutable `canonical_repository_root` and #739/#740 routed
three consumption paths through it. The paths #740 left behind all shared one
shape: the *filesystem* guards had been migrated to the canonical root, but
*repository identity* still bottomed out in `_local_git_remote_url`, which ran
`git remote get-url` with `cwd=PROJECT_ROOT` — always the Gitea-Tools install
checkout. The two halves of a single assessment therefore described two
different repositories.

For a namespace bound to another repository this inverts the guards rather than
merely weakening them: an operation naming the genuinely bound target is
rejected, while one naming Gitea-Tools is accepted.

Central helper
--------------
Adds `_canonical_local_git_root()` as the single place a target-repository root
is resolved: the configured canonical root when one is declared, else
`PROJECT_ROOT`. A configured-but-invalid root is never silently replaced by the
install checkout. Single-repository behaviour is byte-for-byte unchanged.

Defects fixed
-------------
* `_local_git_remote_url` now runs in the canonical target root, which corrects
  every downstream identity consumer at once (`_resolve`, the #530 remote/repo
  guard, the anti-stomp org/repo fill, `_workspace_repository_slug`).
* `_verify_role_mutation_workspace` omitted `configured_canonical_root`, so the
  #274 branches-only / worktree-membership guards validated
  `Gitea-Tools/branches/` instead of the bound target. It now threads the root
  exactly as `_resolve_namespace_mutation_context` does.
* `_resolve` derived omitted coordinates by looking a remote up *by name*. A
  target checkout commonly names its remote `origin` rather than `prgs`, so the
  lookup returned None and the coordinates fell through to the remote-wide
  default target — an unrelated repository. It now prefers
  `_canonical_repository_slug`, which probes candidate remote names.
* Explicit `org`/`repo` short-circuit the #530 match check, so caller
  coordinates bypassed validation entirely. They may now only *confirm* a
  canonical binding, never override it, and fail closed in both directions.
* Reconciler ancestry, fetch, worktree-inventory and cleanup call sites, the
  author worktree derivation in `gitea_lock_issue`/`gitea_create_pr`, and the
  `control_clean` porcelain probe now use the canonical target root.
* `gitea_config`: v2-`environments` silently *dropped* `canonical_repository_root`
  during flattening, so such a namespace fell back to the install root — a
  fail-open. It is now validated and propagated. The v1 path validates it too,
  so all three loaders behave identically.

Preserved as installation-scoped
--------------------------------
Server parity (`master_parity_gate`), workflow/schema/skill loading, self-code
hashing and stale-runtime detection, and `mirror_refs.sh` lookup remain anchored
to `PROJECT_ROOT`. The `server_implementation` vs `target_repository` parity
dimensions from #740 stay separately labelled, and only the server dimension
gates mutations.

Tests
-----
`tests/test_issue_741_canonical_root_consumers.py` (51 tests, 52 subtests) drives
a role-by-operation matrix over author/reviewer/merger/reconciler against:
correct cross-repository target, wrong repository, wrong worktree, explicit
matching and mismatched coordinates, missing/invalid canonical root, immutable
first-bind, and request-override attempts — plus both cross-repository
directions and all three config loaders. Real git repositories, no network, no
branch deletion, no merge.

The module reinstalls the genuine `_local_git_remote_url` per test: an autouse
conftest fixture permanently reassigns it to a working-directory-independent
stub, which is precisely the behaviour under test.

Full suite: 3408 passed, 6 skipped, 2 failed. Both failures
(`test_issue_702_review_findings_f1_f6`, `test_reconciler_supersession_close`)
reproduce identically on clean master c908ed6050 and are pre-existing.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01AYGdWAwuA6UNDc9c3CGipU
2026-07-18 12:02:31 -04:00
sysadmin c908ed6050 Merge pull request 'fix(mcp): complete canonical-root consumption for cross-repository namespaces (Closes #739)' (#740) from feat/issue-739-canonical-root-consumption into master 2026-07-18 10:34:10 -05:00
sysadmin 0a38d92382 Merge pull request 'fix(mcp): accept acquired merger-lease provenance and add owner finalization (Closes #742)' (#743) from fix/issue-742-merger-lease-provenance into master 2026-07-18 10:12:27 -05:00
sysadminandClaude Opus 4.8 15a8a76e99 fix(mcp): complete canonical-root consumption for cross-repo namespaces (Closes #739)
#706 routed the #274 filesystem guards and the session repository slug through
the configured canonical_repository_root. Three consumption paths were left
deriving from the Gitea-Tools installation checkout.

F1 — gitea_get_runtime_context did not normalize its `remote` argument, while
gitea_whoami did. On a prgs-hosted namespace whose first native call was the
runtime-context path, the 'dadeschools' argument default was pinned: identity
resolved against the wrong host, and because first-bind is first-write-wins a
later correct gitea_whoami(remote="prgs") could not repair the binding. It now
calls _effective_remote before any host lookup, identity resolution, or session
seeding. Explicit remotes pass through untouched and a dadeschools-hosted
profile still resolves to dadeschools.

F2 — _delete_branch_repository_binding_block derived its expected slug from
_workspace_repository_slug, which reads _local_git_remote_url in PROJECT_ROOT
(always Gitea-Tools). For a cross-repository namespace this inverted the guard:
the genuinely bound target was rejected and Gitea-Tools was accepted. The
canonical-root branch already present in _trusted_session_repository is
extracted as _canonical_repository_slug and shared by both call sites. A
configured-but-unresolvable root now fails closed instead of falling back to
the installation identity; unconfigured namespaces keep the install-derived
default unchanged.

F3 — gitea_assess_master_parity measures Gitea-Tools server implementation
parity only. That is intentional and is preserved: startup_head, current_head,
in_parity, stale, and restart_required keep their existing meaning and values,
and only the server dimension gates mutations. Two separately labelled
dimensions are added — server_implementation (installation root and commit) and
target_repository (canonical root, repository slug, checkout head, last-known
remote master head, staleness) — so cross-repository commissioning evidence can
distinguish them. The target assessment makes no network call: the remote side
is read from the existing remote-tracking ref, and an unfetched target is
reported indeterminate rather than guessed at.

Verified intentional and left unchanged: the #274 workspace-membership,
author-mutation-worktree, and branches-only guards already validate against the
configured canonical root; explicit org/repo may confirm but never authorize a
binding; and the four-role repository-specific configuration surface already
passes audit and bind-time validation with a wrong root failing closed.

Tests: 31 new cases in tests/test_cross_repo_canonical_consumption.py using
real git repositories and no network. Full suite 3298 passed / 6 skipped
against a clean-baseline 3267; the 2 failures
(test_issue_702_review_findings_f1_f6, test_reconciler_supersession_close)
reproduce identically on unmodified master a8d2087 and are pre-existing.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_013jUxaLVLkPHuTQwAzFzztW
2026-07-18 03:44:28 -04:00
6 changed files with 1984 additions and 53 deletions
+153
View File
@@ -0,0 +1,153 @@
# Installation root vs canonical target repository root
## Purpose
This document (tracked as issue #741, building on #706 and #739/#740) explains
the two distinct filesystem roots the Gitea-Tools MCP server reasons about, why
conflating them silently targets the wrong repository, and which rule applies
when you add a new consumer.
It is the repository-scope companion to
[`gitea-execution-profiles.md`](gitea-execution-profiles.md) (the profile model)
and [`gitea-dual-namespace-deployment.md`](gitea-dual-namespace-deployment.md)
(the per-role namespace model).
## The two roots
| | Installation root | Canonical target repository root |
|---|---|---|
| What it is | The checkout the server *code* lives in | The working root of the repository whose issues/PRs/branches the namespace *mutates* |
| How it is derived | `PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))` | Configured per namespace, then pinned immutably into the session |
| Configured by | Nothing — it follows the script | `canonical_repository_root` profile field, or the `GITEA_CANONICAL_REPOSITORY_ROOT` environment variable |
| Changes at runtime? | No | No — first bind wins for the life of the process |
| Accessor | `PROJECT_ROOT` | `_canonical_local_git_root()` (filesystem) / `_canonical_repository_slug()` (identity) |
For a **single-repository** namespace — every Gitea-Tools namespace today — the
two roots are the same path, and nothing about the existing behaviour changes.
The distinction only becomes observable once a namespace is pointed at a
different repository.
## Which root does my code need?
Ask what the operation is *about*, not where the file happens to sit.
**Use the installation root (`PROJECT_ROOT`)** when the operation concerns the
Gitea-Tools software itself:
- server implementation / version parity (`master_parity_gate`, the
`startup_head` vs `current_head` staleness gate);
- loading the server's own workflow, schema and skill files;
- self-code hashing and stale-runtime detection;
- locating installed scripts such as `mirror_refs.sh`.
These are intentionally install-scoped. Do not "fix" them.
**Use the canonical target root (`_canonical_local_git_root()`)** when the
operation concerns the repository being worked on:
- `git remote get-url` for repository identity;
- branch creation, push, and commit;
- ancestry and merge-base proofs;
- worktree inventory, cleanup, and branch deletion;
- any local git subprocess whose result feeds a mutation guard.
**If you cannot tell, fail closed.** An ambiguous consumer that guesses the
install root is the exact defect class #741 exists to eliminate.
## Why conflating them inverts the guards
Before #741, `_local_git_remote_url()` ran `git remote get-url` with
`cwd=PROJECT_ROOT` unconditionally. Every consumer of repository *identity*
`_resolve`, the #530 remote/repo guard, the anti-stomp org/repo fill,
`_workspace_repository_slug` — therefore read the Gitea-Tools remote and called
it "the workspace", no matter which repository the namespace was bound to.
For a namespace whose canonical root points elsewhere, this **inverts** the
guard rather than merely weakening it:
- an operation naming the genuinely bound target repository is **rejected**,
because that slug does not appear in the Gitea-Tools remote URL;
- an operation naming Gitea-Tools is **accepted**.
The filesystem guards (#274 branches-only and worktree membership) had already
been migrated to the canonical root by #706, so the two halves of a single
assessment described two different repositories.
A related subtlety: repository identity must not be derived by looking a remote
up by *name*. A target checkout commonly names its remote `origin` rather than
`prgs`, so a name-keyed lookup returns nothing and the omitted coordinates fall
through to the remote-wide default *target* — an unrelated repository.
`_canonical_repository_slug()` probes candidate remote names against the
canonical root instead.
`_canonical_local_git_root()` is now the one place a target root is resolved.
Do not re-derive it; new code that needs a target root calls that helper.
## Configuration
Declare the binding on the profile, alongside `allowed_repositories`:
```json
{
"profiles": {
"example-author": {
"role": "author",
"canonical_repository_root": "/absolute/path/to/target-repo",
"allowed_repositories": ["Example-Org/target-repo"]
}
}
}
```
The namespace-scoped environment variable
`GITEA_CANONICAL_REPOSITORY_ROOT` overrides the profile field, and is normally
exported next to the server `cwd` in the MCP client configuration.
Validation is layered, and each layer fails closed:
1. **Config load.** The path must be a non-empty absolute string. All supported
loaders — v1, v2-`environments`, and v2-`contexts` — validate it identically.
(Before #741 only the v2-`contexts` loader validated it, and
v2-`environments` silently *dropped* the field during flattening, so the
namespace fell back to the install root — a fail-open.)
2. **Bind time.** The path must exist, be a git repository, and resolve to a
repository identity matching the session's authorized slug. A configured but
unresolvable root is never replaced by the install identity.
3. **Every mutation.** The pinned root is compared against the live configured
value; a mismatch is treated as a forged or conflicting binding.
`allowed_repositories` remains a separate authorization boundary (#714): the
canonical root determines *which* repository is derived, and
`allowed_repositories` determines whether the session may act on it. A root that
resolves to a repository outside that list fails closed.
## Explicit coordinates confirm, never override
Explicit `org`/`repo` arguments may **confirm** an existing canonical binding.
They can never establish, complete, or replace one. A request naming a
repository that contradicts the binding fails closed, in both directions:
- a Gitea-Tools-rooted namespace cannot mutate another repository;
- a namespace rooted at another repository cannot mutate Gitea-Tools.
This matters because both-explicit coordinates short-circuit the #530
remote/repo match check, so without this rule a caller could name any repository
and skip validation entirely.
No request-supplied workspace, remote, owner, repository, or worktree can
replace the immutable root.
## Parity is reported per dimension
`gitea_assess_master_parity` reports two separately labelled dimensions:
- `server_implementation` — the Gitea-Tools installation checkout. Its
`startup_head` / `current_head` / `stale` / `restart_required` fields keep
their original meaning, and **only this dimension gates mutations**: the
running process executes the code it started with, so a merged fix is not live
until the daemon restarts.
- `target_repository` — the configured canonical target checkout and its
last-known remote master.
"In parity" is a statement about one dimension, never about the whole system.
Read the dimension you actually care about.
+17
View File
@@ -272,6 +272,15 @@ def load_config(path=None):
)
if not isinstance(data.get("profiles"), dict):
raise ConfigError(f"{path} must be a JSON object with a 'profiles' object")
# #741: the v1 path returns `data` unflattened, so nothing else validates
# the cross-repository binding before gitea_auth.get_profile() reads it.
# Validate it here so every supported loader treats the field identically
# (a relative or blank path must never reach the runtime guard).
for _name, _profile in data["profiles"].items():
if isinstance(_profile, dict):
_validate_canonical_repository_root(
_name, _profile.get("canonical_repository_root")
)
return data
@@ -358,6 +367,14 @@ def _flatten_identity(env_name, svc_name, svc, ident_name, ident):
for key in ("role", "username", "execution_profile", "audit_label"):
if ident.get(key):
profile[key] = ident[key]
# #741: the cross-repository binding must survive flattening. Previously
# this key was silently dropped here, so a v2-environments namespace that
# declared canonical_repository_root fell back to the *installation* root
# and mutated Gitea-Tools instead of its target repository — a fail-open.
# Validate it exactly as the v2-contexts loader does before propagating.
_validate_canonical_repository_root(addr, ident.get("canonical_repository_root"))
if ident.get("canonical_repository_root"):
profile["canonical_repository_root"] = ident["canonical_repository_root"]
return addr, profile
+217 -53
View File
@@ -393,6 +393,35 @@ def _configured_canonical_root() -> tuple[str | None, str | None]:
return crr.configured_canonical_root(profile, os.environ)
def _canonical_local_git_root() -> str:
"""Local git working root for *target-repository* operations (#741).
Two distinct roots exist and must never be conflated:
* ``PROJECT_ROOT`` the Gitea-Tools **installation** checkout, i.e. where
this server script lives. Server implementation/version parity, workflow
and skill file loading, and self-code staleness detection are anchored
here and must stay that way.
* the **canonical target repository root** the working root of the
repository whose issues/PRs/branches the namespace actually mutates.
Every local git invocation that reads or proves *target repository* state
(remote identity, ancestry, worktree inventory, cleanup) must run here, or
a cross-repository namespace silently operates on Gitea-Tools instead.
Returns the resolved canonical target root when one is configured, else
``PROJECT_ROOT`` which preserves the single-repository default exactly.
A configured-but-invalid root is *not* silently replaced by the install
checkout: it is returned as-is so downstream git calls fail rather than
succeed against the wrong repository, and
:func:`_enforce_canonical_repository_root` fails the mutation closed first.
"""
configured, _source = _configured_canonical_root()
if not configured:
return PROJECT_ROOT
return crr.resolve_repo_toplevel(configured) or os.path.realpath(configured)
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards.
@@ -1448,6 +1477,12 @@ def _verify_role_mutation_workspace(
git_state = issue_lock_worktree.read_worktree_git_state(
_resolve_preflight_workspace_path(worktree_path)
)
# #741: thread the configured canonical root exactly as
# _resolve_namespace_mutation_context does. Omitting it here made the two
# paths disagree: the #274 branches-only / worktree-membership guards fell
# back to the install checkout and validated Gitea-Tools/branches/ instead
# of the target repository the namespace is actually bound to.
_configured_root, _configured_source = _configured_canonical_root()
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=role,
worktree_path=worktree_path,
@@ -1458,6 +1493,7 @@ def _verify_role_mutation_workspace(
),
profile_name=get_profile().get("profile_name"),
current_branch=git_state.get("current_branch"),
configured_canonical_root=_configured_root,
)
if assessment["block"]:
raise RuntimeError(
@@ -1648,6 +1684,45 @@ def _workspace_repository_slug(remote: str | None) -> str | None:
return session_ctx.format_repository_slug(parsed[0], parsed[1])
def _canonical_repository_slug(
profile: dict,
remote: str | None,
) -> tuple[str | None, list[str]]:
"""Repository identity of the *configured* canonical root, if any (#739 F2).
Returns ``(slug, reasons)``. A namespace with no configured
``canonical_repository_root`` yields ``(None, [])`` so callers keep the
install-derived single-repository default. A configured root that cannot be
resolved to a repository identity yields ``(None, reasons)`` a fail-closed
signal, never a silent fallback to the installation checkout, because that
fallback is exactly what lets a cross-repository namespace validate against
the wrong repository.
Env-over-profile precedence and the existence / git-toplevel / resolvable
remote-identity validation all live in ``crr``.
"""
configured_value, configured_source = crr.configured_canonical_root(
profile, os.environ
)
if not configured_value:
return None, []
assessment = crr.assess_canonical_repository_root(
configured_value=configured_value,
source=configured_source,
expected_slug=None,
process_project_root=PROJECT_ROOT,
remote=remote,
require_binding=True,
)
slug = assessment.get("resolved_slug")
if assessment.get("block") or not slug:
return None, list(assessment.get("reasons") or [
"configured canonical repository root has no resolvable "
"repository identity (fail closed)"
])
return slug, []
def _trusted_session_repository(
profile: dict,
remote: str | None,
@@ -1686,32 +1761,10 @@ def _trusted_session_repository(
# self-authorizing). Env-over-profile precedence and fail-closed validation
# (existence, git toplevel, resolvable remote identity) are handled by
# ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
configured_value, configured_source = crr.configured_canonical_root(
profile, os.environ
)
if configured_value:
assessment = crr.assess_canonical_repository_root(
configured_value=configured_value,
source=configured_source,
expected_slug=None,
process_project_root=PROJECT_ROOT,
remote=remote,
require_binding=True,
)
slug = assessment.get("resolved_slug")
if assessment.get("block") or not slug:
return {
"org": None,
"repository": None,
"reasons": assessment.get("reasons")
or [
"configured canonical repository root has no resolvable "
"repository identity; session repository cannot be "
"authorized (fail closed)"
],
}
else:
slug = _workspace_repository_slug(remote)
canonical_slug, canonical_reasons = _canonical_repository_slug(profile, remote)
if canonical_reasons:
return {"org": None, "repository": None, "reasons": canonical_reasons}
slug = canonical_slug or _workspace_repository_slug(remote)
scope = session_ctx.assess_repository_scope(
workspace_slug=slug,
allowed=allowed,
@@ -2386,9 +2439,24 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
filled_org = False
filled_repo = False
if not org_explicit or not repo_explicit:
parsed = remote_repo_guard.parse_org_repo_from_remote_url(
_local_git_remote_url(remote)
)
parsed = None
# #741: for a cross-repository namespace, prefer the canonical root's
# own repository identity. `_local_git_remote_url` looks a remote up by
# *name*, but a target checkout commonly names its remote `origin`
# rather than `prgs`; that lookup then returns None and the omitted
# coordinates silently fell through to the REMOTES default target —
# i.e. a completely unrelated repository. `_canonical_repository_slug`
# probes the candidate remote names against the canonical root.
try:
canonical_slug, _reasons = _canonical_repository_slug(get_profile(), remote)
except Exception:
canonical_slug = None
if canonical_slug:
parsed = session_ctx.parse_repository_slug(canonical_slug)
if not parsed:
parsed = remote_repo_guard.parse_org_repo_from_remote_url(
_local_git_remote_url(remote)
)
if parsed:
if not org_explicit:
resolved_org = parsed[0]
@@ -2396,6 +2464,40 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
if not repo_explicit:
resolved_repo = parsed[1]
filled_repo = True
# #741: explicit caller coordinates may *confirm* a cross-repository
# canonical binding but must never override it. Both-explicit coordinates
# short-circuit the #530 guard (remote_repo_guard.assess_remote_repo_match),
# so without this check a caller could name any repository and skip
# validation entirely. Unconfigured (single-repository) namespaces yield no
# canonical slug and are unaffected. An unresolvable configured root is not
# blocked here — reads must keep working; the fail-closed decision belongs
# to _enforce_canonical_repository_root at the mutation gate.
if org_explicit or repo_explicit:
try:
canonical_slug, _canonical_reasons = _canonical_repository_slug(
get_profile(), remote
)
except Exception:
canonical_slug = None
bound = (
session_ctx.parse_repository_slug(canonical_slug)
if canonical_slug
else None
)
if bound:
override = session_ctx.assess_repository_override(
requested_org=org,
requested_repo=repo,
bound_org=bound[0],
bound_repo=bound[1],
)
if override.get("block"):
raise RuntimeError(
"Canonical repository binding guard (#741): "
+ "; ".join(override.get("reasons") or [])
+ ". Explicit org/repo may confirm the configured canonical "
"repository binding but can never override it."
)
_enforce_remote_repo_guard(
remote,
resolved_org,
@@ -3050,7 +3152,7 @@ def gitea_lock_issue(
return blocked
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, PROJECT_ROOT
worktree_path, _canonical_local_git_root()
)
h, o, r = _resolve(remote, host, org, repo)
active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
@@ -3352,7 +3454,7 @@ def gitea_create_pr(
locked_worktree = lock_data.get("worktree_path")
worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock(
locked_worktree, worktree_path, PROJECT_ROOT
locked_worktree, worktree_path, _canonical_local_git_root()
)
if worktree_check["block"]:
raise ValueError(worktree_check["reasons"][0])
@@ -8482,13 +8584,24 @@ def gitea_merge_pr(
def _local_git_remote_url(remote_name: str) -> str | None:
"""Best-effort local ``git remote get-url`` for trust-gate corroboration."""
"""Best-effort local ``git remote get-url`` for trust-gate corroboration.
#741: this runs in the **canonical target repository root**, not the
Gitea-Tools installation checkout. Every consumer of this function derives
*target repository identity* from it (``_resolve``, the #530 remote/repo
guard, the anti-stomp org/repo fill, ``_workspace_repository_slug``), so
running it in ``PROJECT_ROOT`` inverted those guards for a cross-repository
namespace: the genuinely bound target was rejected while Gitea-Tools was
accepted. Single-repository namespaces are unaffected
:func:`_canonical_local_git_root` returns ``PROJECT_ROOT`` when no
cross-repository binding is configured.
"""
try:
proc = subprocess.run(
["git", "remote", "get-url", remote_name],
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
cwd=_canonical_local_git_root(),
)
if proc.returncode != 0:
return None
@@ -8768,8 +8881,28 @@ def _delete_branch_repository_binding_block(
(unproven target).
Returns a structured block dict, or ``None`` when the target is authorized.
#739 F2: the trusted identity is the session's configured
``canonical_repository_root`` when the namespace declares one. Deriving it
from ``_workspace_repository_slug`` alone reads the *installation* checkout
(``_local_git_remote_url`` runs in ``PROJECT_ROOT``, always Gitea-Tools),
which inverts the guard for a cross-repository namespace: the genuinely
bound target is rejected and Gitea-Tools is accepted. A configured root that
cannot be resolved fails closed rather than falling back to the install
identity; unconfigured namespaces keep the install-derived default.
"""
workspace_slug = _workspace_repository_slug(remote)
canonical_slug, canonical_reasons = _canonical_repository_slug(
get_profile(), remote
)
if canonical_reasons:
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": canonical_reasons,
"blocker_kind": "repository_binding",
}
workspace_slug = canonical_slug or _workspace_repository_slug(remote)
workspace_parts = (
session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None
)
@@ -9038,7 +9171,7 @@ def gitea_cleanup_merged_pr_branch(
else f"origin/{target_branch}"
)
head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
PROJECT_ROOT,
_canonical_local_git_root(),
head_sha,
target_ref,
)
@@ -9073,7 +9206,7 @@ def gitea_cleanup_merged_pr_branch(
repo=r,
branch=head_branch,
pr_number=pr_number,
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
auth=auth,
base_api=base,
)
@@ -9568,7 +9701,7 @@ def gitea_capture_branches_worktree_snapshot(
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
root = PROJECT_ROOT
root = _canonical_local_git_root()
if worktree_path:
root = os.path.realpath(os.path.abspath(worktree_path))
git_root = _get_git_root(root)
@@ -9678,14 +9811,14 @@ def gitea_reconcile_merged_cleanups(
h, o, r, auth, head_ref
)
head_on_master[int(pr["number"])] = merged_cleanup_reconcile.is_head_ancestor_of_ref(
PROJECT_ROOT, head_sha, master_ref
_canonical_local_git_root(), head_sha, master_ref
)
delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete")
# #534: discover reviewer scratch trees and active leases for those PRs.
scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
PROJECT_ROOT
_canonical_local_git_root()
)
active_reviewer_leases: dict[int, bool] = {}
pr_states: dict[int, dict] = {}
@@ -9711,7 +9844,7 @@ def gitea_reconcile_merged_cleanups(
}
report = merged_cleanup_reconcile.build_reconciliation_report(
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
closed_prs=merged_closed,
open_prs=open_prs,
remote_branch_exists=remote_branch_exists,
@@ -9751,7 +9884,7 @@ def gitea_reconcile_merged_cleanups(
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
auth=auth,
base_api=base,
)
@@ -9835,7 +9968,7 @@ def gitea_reconcile_merged_cleanups(
if local_assessment.get("safe_to_remove_worktree"):
result = merged_cleanup_reconcile.remove_local_worktree(
PROJECT_ROOT,
_canonical_local_git_root(),
head_branch,
worktree_path=local_assessment.get("worktree_path"),
)
@@ -9845,7 +9978,7 @@ def gitea_reconcile_merged_cleanups(
if not scratch.get("safe_to_remove_worktree"):
continue
result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
PROJECT_ROOT, scratch.get("worktree_path") or ""
_canonical_local_git_root(), scratch.get("worktree_path") or ""
)
actions.append({"action": "remove_reviewer_scratch_worktree", **result})
@@ -9943,7 +10076,7 @@ def gitea_assess_already_landed_reconciliation(
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
remote=remote,
target_branch=target_branch,
)
@@ -10039,14 +10172,14 @@ def gitea_scan_already_landed_open_prs(
}
target_fetch = reconciliation_workflow.fetch_target_branch(
PROJECT_ROOT, remote, target_branch
_canonical_local_git_root(), remote, target_branch
)
candidates: list[dict] = []
for pr in open_prs:
assessment = reconciliation_workflow.assess_open_pr_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
remote=remote,
target_branch=target_branch,
target_fetch=target_fetch,
@@ -10143,7 +10276,7 @@ def gitea_audit_worktree_cleanup(
active_issue_branches.add(str(lock["branch_name"]).strip())
report = worktree_cleanup_audit.audit_branches_directory(
PROJECT_ROOT,
_canonical_local_git_root(),
open_pr_branches=open_pr_branches,
active_issue_branches=active_issue_branches,
now=datetime.now(timezone.utc),
@@ -10226,7 +10359,7 @@ def gitea_reconcile_already_landed_pr(
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr=pr,
project_root=PROJECT_ROOT,
project_root=_canonical_local_git_root(),
remote=remote,
target_branch=target_branch,
)
@@ -10431,7 +10564,7 @@ def gitea_reconcile_superseded_by_merged_pr(
)
target_fetch = already_landed_reconcile.fetch_target_branch(
PROJECT_ROOT, remote, target_branch
_canonical_local_git_root(), remote, target_branch
)
superseding_head_sha = (
(superseding_pr.get("head") or {}).get("sha")
@@ -10441,7 +10574,7 @@ def gitea_reconcile_superseded_by_merged_pr(
ancestor = None
if target_fetch.get("success"):
ancestor = already_landed_reconcile.is_head_ancestor_of_ref(
PROJECT_ROOT,
_canonical_local_git_root(),
superseding_head_sha,
target_fetch.get("target_ref") or f"{remote}/{target_branch}",
)
@@ -14226,6 +14359,16 @@ def gitea_get_runtime_context(
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
"""
# #739 F1: normalize the remote *before* any host lookup, identity
# resolution, or session seeding. ``remote`` defaults to 'dadeschools'
# across the whole tool surface, so a caller that omits it on a prgs-hosted
# namespace would otherwise pin the argument default: the identity lookup
# below would authenticate against the wrong host, and because first-bind is
# first-write-wins a later, correct ``gitea_whoami(remote="prgs")`` could not
# repair the binding. ``gitea_whoami`` has always normalized here; this is
# the same call, not a new policy. Explicit non-default remotes pass through
# untouched, and a dadeschools-hosted profile still resolves to dadeschools.
remote = _effective_remote(remote)
profile = get_profile()
config = gitea_config.load_config()
reveal = _reveal_endpoints()
@@ -14403,13 +14546,23 @@ def gitea_assess_master_parity(
Never mutates and makes no network calls. Read-only operations are never
blocked by staleness; only mutating operations fail closed while stale.
The top-level ``startup_head``/``current_head``/``in_parity`` fields keep
their original meaning: the *Gitea-Tools server implementation* only. #739
F3 adds two separately labelled dimensions so a cross-repository namespace
can tell them apart ``server_implementation`` (this installation, restated
under an explicit name) and ``target_repository`` (the configured canonical
target checkout and its last-known remote master). Only the server dimension
gates mutations.
Returns:
dict with 'in_parity', 'stale', 'restart_required', 'startup_head',
'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a
'report' recovery payload when stale.
'current_head', 'mutation_gate_enforced', 'summary', 'reasons',
'server_implementation', 'target_repository', and a 'report' recovery
payload when stale.
"""
parity = _current_master_parity()
enforced = not master_parity_gate.gate_disabled()
canonical_root, canonical_source = _configured_canonical_root()
out = {
"in_parity": parity["in_parity"],
"stale": parity["stale"],
@@ -14421,6 +14574,17 @@ def gitea_assess_master_parity(
"summary": master_parity_gate.format_parity(parity),
"reasons": parity["reasons"],
"process_root": PROJECT_ROOT,
"server_implementation": {
"installation_root": PROJECT_ROOT,
"startup_head": parity["startup_head"],
"current_head": parity["current_head"],
"stale": parity["stale"],
"determinable": parity["determinable"],
},
"target_repository": master_parity_gate.assess_target_repository_parity(
canonical_root=canonical_root,
source=canonical_source,
),
}
if parity["stale"] and enforced:
out["report"] = master_parity_gate.parity_report(parity)
@@ -15667,7 +15831,7 @@ def gitea_update_pr_branch_by_merge(
control_clean = True
try:
porcelain = _get_workspace_porcelain(PROJECT_ROOT)
porcelain = _get_workspace_porcelain(_canonical_local_git_root())
# Untracked-only dirt is ignored; tracked edits contaminate control.
tracked_dirty = [
line for line in (porcelain or "").splitlines()
+125
View File
@@ -166,3 +166,128 @@ def format_parity(assessment: dict) -> str:
if not assessment.get("determinable"):
return "parity indeterminate (baseline or current HEAD unknown)"
return f"in parity at {_short(assessment.get('current_head'))}"
# ---------------------------------------------------------------------------
# Target-repository parity (#739 F3)
#
# Everything above measures ONE dimension: the Gitea-Tools server's own
# implementation commit, comparing the SHA this process was loaded from against
# the SHA now on disk at PROJECT_ROOT. That is deliberate and is left untouched
# — it is what proves the in-memory capability gates are current.
#
# It is not, however, a statement about the repository a cross-repository
# namespace actually mutates. The assessment below is a separate, separately
# labelled dimension for the configured canonical target repository. It never
# feeds the mutation gate and never changes startup_head/current_head.
# ---------------------------------------------------------------------------
DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"
def _git_capture(root: str, *args: str) -> str | None:
"""Run a read-only git command in *root*; ``None`` on any failure.
Deliberately does not honour ``GITEA_TEST_CURRENT_HEAD``: that override
exists to pin the *server's* HEAD, and applying it here would make a target
repository silently report the server's forced SHA.
"""
if not root:
return None
try:
res = subprocess.run(
["git", "-C", root, *args],
capture_output=True,
text=True,
check=False,
)
except Exception:
return None
if res.returncode != 0:
return None
return (res.stdout or "").strip() or None
def assess_target_repository_parity(
*,
canonical_root: str | None,
source: str | None,
tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
) -> dict:
"""Assess the configured cross-repository target checkout.
Reports the target's canonical root, repository identity, checked-out
commit, and last-known remote master commit, plus whether the checkout is
behind that ref. No network call is made: the remote side is read from the
existing remote-tracking ref, so a target that has never been fetched is
reported as indeterminate rather than guessed at.
An unconfigured namespace is ``configured=False`` and never ``stale`` — the
single-repository default has no second dimension to be stale about. A
configured root that cannot be read is ``determinable=False`` with reasons.
"""
result = {
"configured": bool(canonical_root),
"canonical_repository_root": None,
"source": source,
"repository_slug": None,
"checkout_head": None,
"tracking_ref": tracking_ref,
"remote_tracking_head": None,
"determinable": False,
"stale": False,
"reasons": [],
}
if not canonical_root:
result["determinable"] = True
return result
result["canonical_repository_root"] = canonical_root
if not os.path.isdir(canonical_root):
result["reasons"].append(
f"configured canonical repository root '{canonical_root}' "
f"does not exist or is not a directory"
)
return result
toplevel = _git_capture(canonical_root, "rev-parse", "--show-toplevel")
if not toplevel:
result["reasons"].append(
f"configured canonical repository root '{canonical_root}' "
f"is not a git checkout"
)
return result
head = _git_capture(canonical_root, "rev-parse", "HEAD")
if not head:
result["reasons"].append(
f"target repository HEAD could not be read at '{canonical_root}'"
)
return result
result["checkout_head"] = head
result["determinable"] = True
remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
if remote_url:
# Local import keeps this module dependency-light for its startup role.
import remote_repo_guard
parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
if parsed:
result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
if not result["repository_slug"]:
result["reasons"].append(
"target repository identity could not be derived from its git remote"
)
tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
if not tracking_head:
result["reasons"].append(
f"remote-tracking ref '{tracking_ref}' is unknown in the target "
f"checkout; target staleness is indeterminate (no fetch is "
f"performed by this assessment)"
)
return result
result["remote_tracking_head"] = tracking_head
result["stale"] = tracking_head != head
return result
@@ -0,0 +1,704 @@
"""Complete canonical-root consumption for cross-repository MCP namespaces.
#706 introduced an immutable, configured ``canonical_repository_root`` and
routed the #274 filesystem guards and the session repository *slug* through it.
Three consumption paths were left behind, and this module drives each one
through its production entry point:
* **A — remote initialization.** ``gitea_get_runtime_context`` never normalized
its ``remote`` argument, while ``gitea_whoami`` did. A fresh process whose
first native call is the runtime-context path therefore pinned the
``dadeschools`` argument default instead of the profile's configured remote,
and — because first-bind is first-write-wins — no later correct call could
repair it.
* **C — reconciler branch deletion.** The delete-branch repository-binding
guard derived its expected slug from ``_workspace_repository_slug``, which
reads the git remote of the *installation* checkout (always Gitea-Tools),
rather than from the session's configured canonical root.
* **D — parity evidence.** ``gitea_assess_master_parity`` proves Gitea-Tools
*server implementation* parity only, which is intentional. It carried no
target-repository dimension at all, so a cross-repository namespace had no
evidence that its target checkout was current. The existing
``startup_head``/``current_head`` semantics are preserved unchanged and the
target-repository assessment is reported under separately labelled fields.
Groups B and E assert *intentional* behaviour (the #274 guards already consume
the canonical root; the configuration surface already validates a
repository-specific namespace) so that a regression in either is caught.
Real git repositories are used throughout; no network calls are made.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import canonical_repository_root as crr # noqa: E402
import gitea_config # noqa: E402
import gitea_mcp_server as srv # noqa: E402
import master_parity_gate # noqa: E402
import mcp_server # noqa: E402
import namespace_workspace_binding as nwb # noqa: E402
import session_context_binding as session_ctx # noqa: E402
INSTALL_ORG = "Scaled-Tech-Consulting"
INSTALL_REPO = "Gitea-Tools"
INSTALL_SLUG = f"{INSTALL_ORG}/{INSTALL_REPO}"
INSTALL_URL = f"https://gitea.prgs.cc/{INSTALL_SLUG}.git"
TARGET_ORG = "Scaled-Tech-Consulting"
TARGET_REPO = "mcp-control-plane"
TARGET_SLUG = f"{TARGET_ORG}/{TARGET_REPO}"
TARGET_URL = f"https://gitea.prgs.cc/{TARGET_SLUG}.git"
def _git(cwd: str, *args: str) -> str:
res = subprocess.run(
["git", "-C", cwd, *args], capture_output=True, text=True, check=True
)
return res.stdout.strip()
def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
path.mkdir(parents=True, exist_ok=True)
_git(str(path), "init", "-q")
_git(str(path), "config", "user.email", "[email protected]")
_git(str(path), "config", "user.name", "Test")
_git(str(path), "remote", "add", remote_name, remote_url)
(path / "README.md").write_text("seed\n")
_git(str(path), "add", "README.md")
_git(str(path), "commit", "-q", "-m", "seed")
return os.path.realpath(str(path))
# ---------------------------------------------------------------------------
# Shared profile/config construction
# ---------------------------------------------------------------------------
_AUTHOR_OPS = [
"gitea.read",
"gitea.issue.create",
"gitea.issue.comment",
"gitea.pr.create",
"gitea.pr.comment",
"gitea.branch.create",
"gitea.branch.push",
"gitea.repo.commit",
]
_AUTHOR_FORBIDDEN = [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.request_changes",
]
# A profile that may approve or merge must forbid authoring (gitea_config's
# reviewer-identity deadlock rule).
_REVIEWER_OPS = [
"gitea.read",
"gitea.pr.approve",
"gitea.pr.request_changes",
"gitea.pr.comment",
"gitea.issue.comment",
]
_REVIEWER_FORBIDDEN = ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"]
_MERGER_OPS = ["gitea.read", "gitea.pr.merge", "gitea.pr.comment"]
_MERGER_FORBIDDEN = [
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.request_changes",
]
_RECONCILER_OPS = [
"gitea.read",
"gitea.pr.close",
"gitea.pr.comment",
"gitea.branch.delete",
]
_RECONCILER_FORBIDDEN = [
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.request_changes",
]
_ROLE_MATRIX = {
"author": (_AUTHOR_OPS, _AUTHOR_FORBIDDEN),
"reviewer": (_REVIEWER_OPS, _REVIEWER_FORBIDDEN),
"merger": (_MERGER_OPS, _MERGER_FORBIDDEN),
"reconciler": (_RECONCILER_OPS, _RECONCILER_FORBIDDEN),
}
def _profile(
role: str,
*,
canonical_root: str | None = None,
allowed_repositories: list[str] | None = None,
username: str = "jcwalker3",
) -> dict:
allowed, forbidden = _ROLE_MATRIX[role]
profile = {
"enabled": True,
"context": "prgs",
"role": role,
"username": username,
"base_url": "https://gitea.prgs.cc",
"auth": {"type": "env", "name": f"GITEA_TOKEN_PRGS_{role.upper()}"},
"allowed_operations": list(allowed),
"forbidden_operations": list(forbidden),
"execution_profile": f"prgs-{role}",
}
if canonical_root is not None:
profile["canonical_repository_root"] = canonical_root
if allowed_repositories is not None:
profile["allowed_repositories"] = list(allowed_repositories)
return profile
def _config(profiles: dict) -> dict:
return {
"version": 2,
"rules": {"allow_runtime_switching": True},
"contexts": {
"prgs": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
},
"mdcps": {
"enabled": True,
"gitea": {
"enabled": True,
"base_url": "https://gitea.dadeschools.net",
},
},
},
"profiles": profiles,
}
class _ServerHarness(unittest.TestCase):
"""Temp profiles.json + pinned install remote + no network."""
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.tmp = self._dir.name
self.config_path = os.path.join(self.tmp, "profiles.json")
session_ctx._reset_session_context_for_testing()
gitea_config._active_profile_override = None
mcp_server._IDENTITY_CACHE.clear()
srv._MUTATION_AUTHORITY = None
# The install checkout's remote is Gitea-Tools regardless of the
# developer's layout; pin it so "install-derived" is deterministic.
self._remote_url = patch.object(
srv, "_local_git_remote_url", side_effect=self._install_remote_url
)
self._remote_url.start()
def tearDown(self):
self._remote_url.stop()
session_ctx._reset_session_context_for_testing()
gitea_config._active_profile_override = None
mcp_server._IDENTITY_CACHE.clear()
srv._MUTATION_AUTHORITY = None
self._dir.cleanup()
def _install_remote_url(self, remote_name):
return INSTALL_URL if remote_name in ("prgs", "origin") else None
def _write_config(self, profiles: dict) -> None:
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(_config(profiles)))
def _env(self, profile_name: str, **extra) -> dict:
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile_name,
"GITEA_TOKEN_PRGS_AUTHOR": "t",
"GITEA_TOKEN_PRGS_REVIEWER": "t",
"GITEA_TOKEN_PRGS_MERGER": "t",
"GITEA_TOKEN_PRGS_RECONCILER": "t",
}
env.update(extra)
return env
def _api(self, method, url, header):
return {
"login": "jcwalker3",
"full_name": "Test",
"id": 1,
"email": "[email protected]",
}
def _live(self):
return patch("gitea_mcp_server.api_request", side_effect=self._api)
# ===========================================================================
# A. Remote initialization
# ===========================================================================
class TestRemoteInitializationFirstCall(_ServerHarness):
"""A prgs namespace must never pin the dadeschools argument default."""
def setUp(self):
super().setUp()
self._write_config(
{"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
)
def test_runtime_context_first_call_does_not_bind_default_remote(self):
"""Runtime-context as the FIRST native call, relying on the default."""
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
srv.gitea_get_runtime_context()
ctx = session_ctx.get_session_context()
self.assertIsNotNone(ctx, "first call must establish a session binding")
self.assertEqual(ctx["remote"], "prgs")
self.assertEqual(ctx["host"], "gitea.prgs.cc")
self.assertEqual(ctx["org"], INSTALL_ORG)
self.assertEqual(ctx["repository"], INSTALL_REPO)
def test_runtime_context_first_call_reports_effective_remote(self):
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
result = srv.gitea_get_runtime_context()
self.assertEqual(result["remote"], "prgs")
def test_whoami_first_call_remains_correct(self):
"""Control: the already-normalizing path is unchanged."""
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
srv.gitea_whoami()
ctx = session_ctx.get_session_context()
self.assertEqual(ctx["remote"], "prgs")
self.assertEqual(ctx["host"], "gitea.prgs.cc")
def test_explicit_remote_argument_still_honoured(self):
"""Normalization only fills the default; explicit values are untouched."""
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
result = srv.gitea_get_runtime_context(remote="prgs")
self.assertEqual(result["remote"], "prgs")
def test_mdcps_profile_default_is_not_rewritten_to_prgs(self):
"""A dadeschools-hosted profile keeps the dadeschools remote."""
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
profile["context"] = "mdcps"
profile["base_url"] = "https://gitea.dadeschools.net"
self._write_config({"prgs-author": profile})
with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
result = srv.gitea_get_runtime_context()
self.assertEqual(result["remote"], "dadeschools")
# ===========================================================================
# B. Remote/repository guard (intentional behaviour — regression fence)
# ===========================================================================
class TestCanonicalRootGuardBinding(_ServerHarness):
def setUp(self):
super().setUp()
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
self.install_root = _init_repo(Path(self.tmp) / "install", INSTALL_URL)
def test_canonical_target_root_resolves_to_target_slug(self):
got = crr.assess_canonical_repository_root(
configured_value=self.target_root,
source="profile.canonical_repository_root",
expected_slug=None,
process_project_root=self.install_root,
remote="prgs",
require_binding=True,
)
self.assertFalse(got.get("block"), got.get("reasons"))
self.assertEqual(got["resolved_slug"], TARGET_SLUG)
def test_install_root_identity_remains_gitea_tools(self):
got = crr.assess_canonical_repository_root(
configured_value=None,
source=None,
expected_slug=None,
process_project_root=self.install_root,
remote="prgs",
)
self.assertEqual(
os.path.realpath(got["canonical_repo_root"]), self.install_root
)
def test_guards_validate_against_canonical_target_root(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="author",
worktree_path=None,
process_project_root=self.install_root,
configured_canonical_root=self.target_root,
)
self.assertEqual(
os.path.realpath(ctx["canonical_repo_root"]), self.target_root
)
def test_explicit_coordinates_cannot_override_canonical_binding(self):
"""Explicit org/repo may confirm, never authorize, a binding."""
confirm = session_ctx.assess_repository_override(
requested_org=TARGET_ORG,
requested_repo=TARGET_REPO,
bound_org=TARGET_ORG,
bound_repo=TARGET_REPO,
)
self.assertFalse(confirm.get("block"))
override = session_ctx.assess_repository_override(
requested_org=TARGET_ORG,
requested_repo=TARGET_REPO,
bound_org=INSTALL_ORG,
bound_repo=INSTALL_REPO,
)
self.assertTrue(override.get("block"))
def test_gitea_tools_rooted_namespace_cannot_reach_target_repository(self):
"""No canonical root configured → session stays pinned to Gitea-Tools."""
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
with patch.dict(os.environ, {}, clear=False):
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
resolved = srv._trusted_session_repository(
profile, "prgs", for_mutation=True
)
self.assertEqual(resolved["repository"], INSTALL_REPO)
self.assertNotEqual(resolved["repository"], TARGET_REPO)
def test_target_rooted_namespace_binds_target_repository(self):
profile = _profile(
"author",
canonical_root=self.target_root,
allowed_repositories=[TARGET_SLUG],
)
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
self.assertEqual(resolved["org"], TARGET_ORG)
self.assertEqual(resolved["repository"], TARGET_REPO)
# ===========================================================================
# C. Reconciler branch deletion
# ===========================================================================
class TestDeleteBranchRepositoryBinding(_ServerHarness):
"""The binding guard must consult the canonical root, not PROJECT_ROOT.
No branch is ever deleted here: only the pre-deletion binding guard is
exercised.
"""
def setUp(self):
super().setUp()
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
self._write_config(
{
"prgs-reconciler": _profile(
"reconciler",
canonical_root=self.target_root,
allowed_repositories=[TARGET_SLUG],
),
"gt-reconciler": _profile(
"reconciler", allowed_repositories=[INSTALL_SLUG]
),
}
)
def test_target_rooted_reconciler_validates_target_deletion(self):
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
block = srv._delete_branch_repository_binding_block(
"prgs", org=TARGET_ORG, repo=TARGET_REPO
)
self.assertIsNone(block, block)
def test_target_rooted_reconciler_fails_closed_for_install_repository(self):
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
block = srv._delete_branch_repository_binding_block(
"prgs", org=INSTALL_ORG, repo=INSTALL_REPO
)
self.assertIsNotNone(block)
self.assertEqual(block["blocker_kind"], "repository_binding")
self.assertFalse(block["performed"])
def test_install_rooted_reconciler_fails_closed_for_target_repository(self):
with patch.dict(os.environ, self._env("gt-reconciler"), clear=False):
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
block = srv._delete_branch_repository_binding_block(
"prgs", org=TARGET_ORG, repo=TARGET_REPO
)
self.assertIsNotNone(block)
self.assertEqual(block["blocker_kind"], "repository_binding")
def test_env_configured_canonical_root_is_honoured(self):
env = self._env("gt-reconciler")
env[crr.CANONICAL_ROOT_ENV] = self.target_root
with patch.dict(os.environ, env, clear=False):
allowed = srv._delete_branch_repository_binding_block(
"prgs", org=TARGET_ORG, repo=TARGET_REPO
)
blocked = srv._delete_branch_repository_binding_block(
"prgs", org=INSTALL_ORG, repo=INSTALL_REPO
)
self.assertIsNone(allowed, allowed)
self.assertIsNotNone(blocked)
def test_unresolvable_canonical_root_fails_closed(self):
env = self._env("gt-reconciler")
env[crr.CANONICAL_ROOT_ENV] = os.path.join(self.tmp, "does-not-exist")
with patch.dict(os.environ, env, clear=False):
block = srv._delete_branch_repository_binding_block(
"prgs", org=TARGET_ORG, repo=TARGET_REPO
)
self.assertIsNotNone(block)
self.assertEqual(block["blocker_kind"], "repository_binding")
def test_no_request_parameter_can_override_canonical_identity(self):
"""Every explicit coordinate that is not the canonical target is
refused; none of them can *establish* the binding."""
# Both repositories share an org, so a wrong *org* case must name a
# genuinely different owner to be meaningful.
cases = [
(INSTALL_ORG, INSTALL_REPO),
(TARGET_ORG, INSTALL_REPO),
(TARGET_ORG, "some-other-repo"),
("someone-else", TARGET_REPO),
]
with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
for org, repo in cases:
with self.subTest(org=org, repo=repo):
block = srv._delete_branch_repository_binding_block(
"prgs", org=org, repo=repo
)
self.assertIsNotNone(block, f"{org}/{repo} must fail closed")
# ===========================================================================
# D. Parity semantics
# ===========================================================================
class TestTargetRepositoryParityAssessment(unittest.TestCase):
"""Server-implementation parity is preserved; target parity is additive."""
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.tmp = self._dir.name
def tearDown(self):
self._dir.cleanup()
def _target(self) -> str:
return _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
def test_unconfigured_target_is_reported_not_stale(self):
got = master_parity_gate.assess_target_repository_parity(
canonical_root=None, source=None
)
self.assertFalse(got["configured"])
self.assertFalse(got["stale"])
self.assertIsNone(got["canonical_repository_root"])
def test_configured_target_reports_checkout_head_and_slug(self):
root = self._target()
head = _git(root, "rev-parse", "HEAD")
got = master_parity_gate.assess_target_repository_parity(
canonical_root=root, source="profile.canonical_repository_root"
)
self.assertTrue(got["configured"])
self.assertEqual(got["canonical_repository_root"], root)
self.assertEqual(got["checkout_head"], head)
self.assertEqual(got["repository_slug"], TARGET_SLUG)
def test_target_stale_when_remote_tracking_ref_is_ahead(self):
root = self._target()
head = _git(root, "rev-parse", "HEAD")
# Simulate a fetched remote-tracking ref that has advanced.
_git(root, "checkout", "-q", "-b", "advanced")
(Path(root) / "next.md").write_text("next\n")
_git(root, "add", "next.md")
_git(root, "commit", "-q", "-m", "advance")
advanced = _git(root, "rev-parse", "HEAD")
_git(root, "update-ref", "refs/remotes/origin/master", advanced)
_git(root, "checkout", "-q", "--detach", head)
got = master_parity_gate.assess_target_repository_parity(
canonical_root=root, source="profile.canonical_repository_root"
)
self.assertEqual(got["checkout_head"], head)
self.assertEqual(got["remote_tracking_head"], advanced)
self.assertTrue(got["stale"])
def test_missing_root_is_not_determinable_and_fails_closed(self):
got = master_parity_gate.assess_target_repository_parity(
canonical_root=os.path.join(self.tmp, "absent"),
source="profile.canonical_repository_root",
)
self.assertTrue(got["configured"])
self.assertFalse(got["determinable"])
self.assertTrue(got["reasons"])
class TestParityToolEvidenceDimensions(_ServerHarness):
def setUp(self):
super().setUp()
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
self._write_config(
{
"prgs-author": _profile(
"author",
canonical_root=self.target_root,
allowed_repositories=[TARGET_SLUG],
)
}
)
def test_existing_server_parity_semantics_are_unchanged(self):
with patch.dict(os.environ, self._env("prgs-author"), clear=False):
result = srv.gitea_assess_master_parity(remote="prgs")
# startup_head/current_head keep their Gitea-Tools implementation
# meaning and must not be redefined to the target repository.
self.assertEqual(result["process_root"], srv.PROJECT_ROOT)
self.assertEqual(
result["startup_head"], srv._STARTUP_PARITY.get("startup_head")
)
self.assertIn("in_parity", result)
def test_evidence_distinguishes_server_and_target_dimensions(self):
with patch.dict(os.environ, self._env("prgs-author"), clear=False):
result = srv.gitea_assess_master_parity(remote="prgs")
server = result["server_implementation"]
self.assertEqual(server["installation_root"], srv.PROJECT_ROOT)
self.assertEqual(server["current_head"], result["current_head"])
self.assertIn("stale", server)
target = result["target_repository"]
self.assertTrue(target["configured"])
self.assertEqual(target["canonical_repository_root"], self.target_root)
self.assertEqual(target["repository_slug"], TARGET_SLUG)
self.assertEqual(
target["checkout_head"], _git(self.target_root, "rev-parse", "HEAD")
)
self.assertIn("stale", target)
def test_unconfigured_namespace_reports_target_as_unconfigured(self):
self._write_config(
{"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
)
env = self._env("prgs-author")
with patch.dict(os.environ, env, clear=False):
os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
result = srv.gitea_assess_master_parity(remote="prgs")
self.assertFalse(result["target_repository"]["configured"])
# ===========================================================================
# E. Startup / configuration validation for a candidate namespace set
# ===========================================================================
class TestCandidateNamespaceConfiguration(_ServerHarness):
"""Four repository-specific profiles for the target repository."""
def setUp(self):
super().setUp()
self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
self.profiles = {
f"mcpcp-{role}": _profile(
role,
canonical_root=self.target_root,
allowed_repositories=[TARGET_SLUG],
)
for role in ("author", "reviewer", "merger", "reconciler")
}
self._write_config(self.profiles)
def test_configuration_audit_succeeds(self):
with patch.dict(os.environ, self._env("mcpcp-author"), clear=False):
audit = srv.gitea_audit_config()
self.assertTrue(audit.get("configured"))
names = {row["name"] for row in audit["profiles"]}
self.assertEqual(names, set(self.profiles))
def test_no_inline_credentials_are_present(self):
raw = json.loads(Path(self.config_path).read_text())
for name, profile in raw["profiles"].items():
with self.subTest(profile=name):
self.assertNotIn("token", profile)
self.assertNotIn("password", profile)
self.assertNotIn("secret", profile)
self.assertIn(profile["auth"]["type"], ("env", "keychain"))
def test_bind_time_validation_succeeds_for_target_repository(self):
for name, profile in self.profiles.items():
with self.subTest(profile=name):
resolved = srv._trusted_session_repository(
profile, "prgs", for_mutation=True
)
self.assertEqual(resolved["reasons"], [])
self.assertEqual(resolved["org"], TARGET_ORG)
self.assertEqual(resolved["repository"], TARGET_REPO)
def test_wrong_repository_root_fails_closed(self):
wrong_root = _init_repo(Path(self.tmp) / "wrong", INSTALL_URL)
profile = _profile(
"author", canonical_root=wrong_root, allowed_repositories=[TARGET_SLUG]
)
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
self.assertIsNone(resolved["repository"])
self.assertTrue(resolved["reasons"])
def test_existing_install_profiles_are_unchanged_in_behaviour(self):
profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
self.assertEqual(resolved["org"], INSTALL_ORG)
self.assertEqual(resolved["repository"], INSTALL_REPO)
def _denied(self, profile: dict, operation: str) -> bool:
ok, _ = gitea_config.check_operation(
operation,
profile["allowed_operations"],
profile["forbidden_operations"],
)
return not ok
def test_role_separation_is_enforced(self):
expectations = {
"author": [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.request_changes",
],
"reviewer": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
"merger": [
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.request_changes",
],
"reconciler": [
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.request_changes",
],
}
for role, denied_ops in expectations.items():
profile = self.profiles[f"mcpcp-{role}"]
for op in denied_ops:
with self.subTest(role=role, operation=op):
self.assertTrue(
self._denied(profile, op),
f"{role} must not be permitted {op}",
)
def test_each_role_retains_its_own_capability(self):
permitted = {
"author": "gitea.pr.create",
"reviewer": "gitea.pr.approve",
"merger": "gitea.pr.merge",
"reconciler": "gitea.branch.delete",
}
for role, op in permitted.items():
with self.subTest(role=role, operation=op):
self.assertFalse(self._denied(self.profiles[f"mcpcp-{role}"], op))
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,768 @@
"""Complete PROJECT_ROOT elimination in cross-repository MCP operations (#741).
#706 introduced the immutable ``canonical_repository_root``; #739/#740 routed
three consumption paths through it. This module covers the paths #740 left
behind, whose shape is uniform: the *filesystem* guards were migrated to the
canonical root, but *repository identity* still bottomed out in
``_local_git_remote_url``'s hardcoded ``cwd=PROJECT_ROOT`` — always the
Gitea-Tools installation checkout.
The consequences asserted here:
* **Identity inversion.** ``_resolve``, the #530 remote/repo guard and the
anti-stomp org/repo fill all derived the *target* repository from the
*install* checkout, so a cross-repository namespace resolved Gitea-Tools
coordinates while its branch/parity facts came from the target repo.
* **Guard disagreement.** ``_verify_role_mutation_workspace`` omitted
``configured_canonical_root``, so the #274 branches-only / worktree-membership
guards validated ``Gitea-Tools/branches/`` rather than the bound target.
* **Explicit-coordinate override.** Both-explicit ``org``/``repo`` short-circuit
``assess_remote_repo_match``, so caller coordinates bypassed validation
entirely rather than merely *confirming* the binding.
* **Configuration fail-open.** ``_flatten_identity`` silently dropped
``canonical_repository_root``, so a v2-``environments`` namespace fell back to
the install root; the v1 path never validated the field at all.
A role-by-operation matrix drives author, reviewer, merger and reconciler
against: correct cross-repository target, wrong repository, wrong worktree,
explicit matching coordinates, explicit mismatched coordinates, missing
canonical root, invalid canonical root, immutable binding after first bind, and
request-override attempts.
Real git repositories are used throughout. No network calls are made, no branch
is deleted, and no merge is performed.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import canonical_repository_root as crr # noqa: E402
import gitea_config # noqa: E402
import gitea_mcp_server as srv # noqa: E402
import mcp_server # noqa: E402
import session_context_binding as session_ctx # noqa: E402
INSTALL_ORG = "Scaled-Tech-Consulting"
INSTALL_REPO = "Gitea-Tools"
INSTALL_SLUG = f"{INSTALL_ORG}/{INSTALL_REPO}"
INSTALL_URL = f"https://gitea.prgs.cc/{INSTALL_SLUG}.git"
TARGET_ORG = "Scaled-Tech-Consulting"
TARGET_REPO = "mcp-control-plane"
TARGET_SLUG = f"{TARGET_ORG}/{TARGET_REPO}"
TARGET_URL = f"https://gitea.prgs.cc/{TARGET_SLUG}.git"
THIRD_REPO = "Timesheet"
THIRD_SLUG = f"{INSTALL_ORG}/{THIRD_REPO}"
# tests/conftest.py installs an autouse fixture
# (mutation_profile_fixture.install_deterministic_remote_urls) that *permanently
# reassigns* srv._local_git_remote_url to a stub mapping remote names to fixed
# URLs. That stub deliberately ignores the working directory, which is exactly
# the behaviour this module must verify — so these tests would silently assert
# against the stub rather than production code in a full-suite run. Capture the
# genuine implementation at import time (before any fixture executes) and
# reinstall it per test.
_REAL_LOCAL_GIT_REMOTE_URL = srv._local_git_remote_url
def _git(cwd: str, *args: str) -> str:
res = subprocess.run(
["git", "-C", cwd, *args], capture_output=True, text=True, check=True
)
return res.stdout.strip()
def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
path.mkdir(parents=True, exist_ok=True)
_git(str(path), "init", "-q")
_git(str(path), "config", "user.email", "[email protected]")
_git(str(path), "config", "user.name", "Test")
_git(str(path), "remote", "add", remote_name, remote_url)
# Distinct content per repo: identical seed content, author and timestamp
# otherwise produce byte-identical commits and therefore an identical SHA,
# which would make the parity-dimension assertions vacuously true.
(path / "README.md").write_text(f"seed {remote_url}\n")
_git(str(path), "add", "README.md")
_git(str(path), "commit", "-q", "-m", f"seed {remote_name}")
return os.path.realpath(str(path))
_ROLE_OPS = {
"author": (
[
"gitea.read",
"gitea.issue.create",
"gitea.issue.comment",
"gitea.pr.create",
"gitea.pr.comment",
"gitea.branch.create",
"gitea.branch.push",
"gitea.repo.commit",
],
["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
),
"reviewer": (
[
"gitea.read",
"gitea.pr.approve",
"gitea.pr.request_changes",
"gitea.pr.comment",
"gitea.issue.comment",
],
["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
),
"merger": (
["gitea.read", "gitea.pr.merge", "gitea.pr.comment"],
[
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.request_changes",
],
),
"reconciler": (
["gitea.read", "gitea.pr.close", "gitea.pr.comment", "gitea.branch.delete"],
[
"gitea.pr.create",
"gitea.branch.push",
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.request_changes",
],
),
}
ROLES = tuple(_ROLE_OPS)
def _profile(role: str, *, canonical_root: str | None = None) -> dict:
allowed, forbidden = _ROLE_OPS[role]
profile = {
"enabled": True,
"context": "prgs",
"role": role,
"username": "jcwalker3",
"base_url": "https://gitea.prgs.cc",
"auth": {"type": "env", "name": f"GITEA_TOKEN_PRGS_{role.upper()}"},
"allowed_operations": list(allowed),
"forbidden_operations": list(forbidden),
"execution_profile": f"prgs-{role}",
"allowed_repositories": [TARGET_SLUG],
}
if canonical_root is not None:
profile["canonical_repository_root"] = canonical_root
return profile
def _config(profiles: dict) -> dict:
return {
"version": 2,
"rules": {"allow_runtime_switching": True},
"contexts": {
"prgs": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
}
},
"profiles": profiles,
}
class _CrossRepoHarness(unittest.TestCase):
"""Real install + target git repos, temp profiles.json, no network."""
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.tmp = self._dir.name
self.config_path = os.path.join(self.tmp, "profiles.json")
# The real install checkout carries a remote literally named `prgs`;
# a freshly cloned target repository normally names its remote `origin`.
# Reproducing that asymmetry is the point: identity derivation must not
# depend on the remote happening to share the `remote=` argument's name.
self.install_root = _init_repo(
Path(self.tmp) / "install", INSTALL_URL, remote_name="prgs"
)
self.target_root = _init_repo(
Path(self.tmp) / "target", TARGET_URL, remote_name="origin"
)
self.third_root = _init_repo(
Path(self.tmp) / "third", f"https://gitea.prgs.cc/{THIRD_SLUG}.git"
)
self.not_a_repo = os.path.join(self.tmp, "plain-dir")
os.makedirs(self.not_a_repo, exist_ok=True)
session_ctx._reset_session_context_for_testing()
gitea_config._active_profile_override = None
mcp_server._IDENTITY_CACHE.clear()
srv._MUTATION_AUTHORITY = None
# PROJECT_ROOT is wherever the server file physically lives; pin it to a
# real Gitea-Tools-identified checkout so "install-derived" is
# deterministic regardless of the developer's layout.
self._project_root = patch.object(srv, "PROJECT_ROOT", self.install_root)
self._project_root.start()
# Undo the autouse deterministic-remote stub for this module only.
self._real_remote_url = patch.object(
srv, "_local_git_remote_url", _REAL_LOCAL_GIT_REMOTE_URL
)
self._real_remote_url.start()
def tearDown(self):
self._real_remote_url.stop()
self._project_root.stop()
session_ctx._reset_session_context_for_testing()
gitea_config._active_profile_override = None
mcp_server._IDENTITY_CACHE.clear()
srv._MUTATION_AUTHORITY = None
self._dir.cleanup()
def _write_config(self, profiles: dict) -> None:
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(_config(profiles)))
def _env(self, role: str, **extra) -> dict:
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": f"prgs-{role}",
"GITEA_TOKEN_PRGS_AUTHOR": "t",
"GITEA_TOKEN_PRGS_REVIEWER": "t",
"GITEA_TOKEN_PRGS_MERGER": "t",
"GITEA_TOKEN_PRGS_RECONCILER": "t",
}
env.update(extra)
return env
def _bind(self, role: str, canonical_root: str | None):
"""Activate *role* with *canonical_root* and return an env patch ctx."""
self._write_config(
{f"prgs-{role}": _profile(role, canonical_root=canonical_root)}
)
return patch.dict(os.environ, self._env(role), clear=True)
# ===========================================================================
# 1. The central helper (single source of root resolution)
# ===========================================================================
class TestCanonicalLocalGitRoot(_CrossRepoHarness):
"""_canonical_local_git_root is the one place a target root is derived."""
def test_unconfigured_namespace_uses_installation_root(self):
with self._bind("author", None):
self.assertEqual(srv._canonical_local_git_root(), self.install_root)
def test_configured_namespace_uses_target_root(self):
with self._bind("author", self.target_root):
self.assertEqual(srv._canonical_local_git_root(), self.target_root)
def test_configured_root_resolves_to_git_toplevel(self):
nested = os.path.join(self.target_root, "branches", "wt")
os.makedirs(nested, exist_ok=True)
with self._bind("author", nested):
# A subdirectory of the target repo still resolves to its toplevel.
self.assertEqual(srv._canonical_local_git_root(), self.target_root)
def test_invalid_root_never_silently_becomes_installation_root(self):
"""A configured-but-broken root must not fall back to Gitea-Tools."""
with self._bind("author", self.not_a_repo):
resolved = srv._canonical_local_git_root()
self.assertNotEqual(resolved, self.install_root)
self.assertEqual(resolved, os.path.realpath(self.not_a_repo))
def test_env_override_beats_profile_binding(self):
self._write_config(
{"prgs-author": _profile("author", canonical_root=self.third_root)}
)
env = self._env("author", GITEA_CANONICAL_REPOSITORY_ROOT=self.target_root)
with patch.dict(os.environ, env, clear=True):
self.assertEqual(srv._canonical_local_git_root(), self.target_root)
# ===========================================================================
# 2. Repository identity — the upstream inversion
# ===========================================================================
class TestRepositoryIdentityDerivation(_CrossRepoHarness):
"""_local_git_remote_url and its consumers must read the target repo."""
def test_remote_url_reads_target_not_installation(self):
with self._bind("author", self.target_root):
self.assertEqual(srv._local_git_remote_url("origin"), TARGET_URL)
def test_remote_url_unconfigured_still_reads_installation(self):
with self._bind("author", None):
# The install checkout's remote is named `prgs`, matching production.
self.assertEqual(srv._local_git_remote_url("prgs"), INSTALL_URL)
def test_workspace_slug_follows_canonical_root(self):
with self._bind("author", self.target_root):
self.assertEqual(srv._workspace_repository_slug("origin"), TARGET_SLUG)
def test_workspace_slug_unconfigured_is_installation(self):
with self._bind("author", None):
self.assertEqual(srv._workspace_repository_slug("prgs"), INSTALL_SLUG)
def test_every_role_derives_the_same_target_identity(self):
for role in ROLES:
with self.subTest(role=role):
with self._bind(role, self.target_root):
self.assertEqual(
srv._workspace_repository_slug("origin"), TARGET_SLUG
)
# ===========================================================================
# 3. _resolve — omitted and explicit coordinates
# ===========================================================================
class TestResolveTargetCoordinates(_CrossRepoHarness):
"""Omitted coordinates follow the binding; explicit ones may only confirm."""
def test_omitted_coordinates_resolve_to_target_repository(self):
for role in ROLES:
with self.subTest(role=role):
with self._bind(role, self.target_root):
_host, org, repo = srv._resolve("prgs", None, None, None)
self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
def test_omitted_coordinates_unconfigured_resolve_to_installation(self):
with self._bind("author", None):
_host, org, repo = srv._resolve("prgs", None, None, None)
self.assertEqual((org, repo), (INSTALL_ORG, INSTALL_REPO))
def test_explicit_matching_coordinates_confirm_the_binding(self):
for role in ROLES:
with self.subTest(role=role):
with self._bind(role, self.target_root):
_host, org, repo = srv._resolve(
"prgs", None, TARGET_ORG, TARGET_REPO
)
self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
def test_explicit_mismatched_repository_cannot_override_binding(self):
for role in ROLES:
with self.subTest(role=role):
with self._bind(role, self.target_root):
with self.assertRaises(RuntimeError) as ctx:
srv._resolve("prgs", None, TARGET_ORG, INSTALL_REPO)
msg = str(ctx.exception)
self.assertIn("#741", msg)
self.assertIn("fail closed", msg)
def test_explicit_mismatched_organization_cannot_override_binding(self):
with self._bind("author", self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, "SomeoneElse", TARGET_REPO)
def test_gitea_tools_rooted_namespace_cannot_target_control_plane(self):
"""Direction 1: an install-rooted namespace must not reach the target."""
with self._bind("author", self.install_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, TARGET_ORG, TARGET_REPO)
def test_control_plane_rooted_namespace_cannot_target_gitea_tools(self):
"""Direction 2: a target-rooted namespace must not reach Gitea-Tools."""
with self._bind("author", self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
def test_unconfigured_namespace_keeps_existing_explicit_behaviour(self):
"""Same-repository behaviour is unchanged (no new fail-closed path)."""
with self._bind("author", None):
_host, org, repo = srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
self.assertEqual((org, repo), (INSTALL_ORG, INSTALL_REPO))
# ===========================================================================
# 4. #274 workspace guard — the two guard paths must agree
# ===========================================================================
class TestMutationWorkspaceGuardBinding(_CrossRepoHarness):
"""Both guard paths must agree on which repository they protect."""
def _contexts(self, worktree: str | None = None):
return srv._resolve_namespace_mutation_context(worktree)
def test_mutation_context_uses_target_root(self):
with self._bind("author", self.target_root):
self.assertEqual(self._contexts()["canonical_repo_root"], self.target_root)
def test_mutation_context_unconfigured_uses_installation_root(self):
with self._bind("author", None):
self.assertEqual(self._contexts()["canonical_repo_root"], self.install_root)
def test_role_mutation_workspace_guard_agrees_with_mutation_context(self):
"""The omitted configured_canonical_root made these two disagree."""
import namespace_workspace_binding as nwb
for role in ROLES:
with self.subTest(role=role):
with self._bind(role, self.target_root):
configured, _src = srv._configured_canonical_root()
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=role,
worktree_path=None,
worktree=None,
process_project_root=srv.PROJECT_ROOT,
profile_name=f"prgs-{role}",
configured_canonical_root=configured,
)
self.assertEqual(
assessment["canonical_repo_root"],
self._contexts()["canonical_repo_root"],
)
self.assertEqual(
assessment["canonical_repo_root"], self.target_root
)
def test_wrong_worktree_is_rejected_against_target_root(self):
"""A worktree belonging to the install repo is not in the target repo."""
import author_mutation_worktree as amw
with self._bind("author", self.target_root):
foreign = os.path.join(self.install_root, "branches", "wt")
os.makedirs(foreign, exist_ok=True)
membership = amw.assess_workspace_repo_membership(
workspace_path=foreign,
canonical_repo_root=self.target_root,
)
self.assertTrue(membership["block"])
# ===========================================================================
# 5. Canonical-root validation — missing / invalid / mismatched
# ===========================================================================
class TestCanonicalRootFailClosed(_CrossRepoHarness):
"""Missing, invalid, ambiguous and mismatched roots fail closed."""
def _assess(self, value, *, expected=TARGET_SLUG, require=True):
return crr.assess_canonical_repository_root(
configured_value=value,
source="test",
expected_slug=expected,
process_project_root=self.install_root,
remote="origin",
require_binding=require,
)
def test_missing_root_fails_closed_when_binding_required(self):
result = self._assess(None)
self.assertTrue(result["block"])
self.assertFalse(result["configured"])
def test_missing_root_is_the_single_repo_default_when_not_required(self):
result = self._assess(None, expected=None, require=False)
self.assertFalse(result["block"])
self.assertEqual(result["canonical_repo_root"], self.install_root)
def test_nonexistent_root_fails_closed(self):
result = self._assess(os.path.join(self.tmp, "nope"))
self.assertTrue(result["block"])
self.assertIn("does not exist", " ".join(result["reasons"]))
def test_non_git_directory_fails_closed(self):
result = self._assess(self.not_a_repo)
self.assertTrue(result["block"])
self.assertIn("not a git repository", " ".join(result["reasons"]))
def test_mismatched_repository_identity_fails_closed(self):
result = self._assess(self.third_root)
self.assertTrue(result["block"])
self.assertIn("mismatch", " ".join(result["reasons"]))
def test_matching_repository_identity_is_proven(self):
result = self._assess(self.target_root)
self.assertFalse(result["block"])
self.assertEqual(result["resolved_slug"], TARGET_SLUG)
self.assertEqual(result["canonical_repo_root"], self.target_root)
def test_unresolvable_root_yields_fail_closed_reasons_not_fallback(self):
with self._bind("author", self.not_a_repo):
slug, reasons = srv._canonical_repository_slug(srv.get_profile(), "origin")
self.assertIsNone(slug)
self.assertTrue(reasons)
# ===========================================================================
# 6. Immutability — first bind wins, requests cannot replace the root
# ===========================================================================
class TestCanonicalRootImmutability(_CrossRepoHarness):
"""No request-supplied value can establish or swap the pinned root."""
def test_first_bind_pins_target_root(self):
with self._bind("author", self.target_root):
with patch(
"gitea_mcp_server.api_request",
side_effect=lambda *a, **k: {
"login": "jcwalker3",
"full_name": "T",
"id": 1,
"email": "[email protected]",
},
):
srv.gitea_whoami(remote="prgs")
bound = session_ctx.get_session_context()
self.assertEqual(bound["canonical_repository_root"], self.target_root)
def test_binding_is_first_write_wins(self):
with self._bind("author", self.target_root):
session_ctx.seed_session_context_if_unbound(
profile_name="prgs-author",
remote="prgs",
host="gitea.prgs.cc",
identity="jcwalker3",
org=TARGET_ORG,
repository=TARGET_REPO,
role_kind="author",
source="test",
canonical_repository_root=self.target_root,
)
# A second, contradictory seed must not replace the pin.
session_ctx.seed_session_context_if_unbound(
profile_name="prgs-author",
remote="prgs",
host="gitea.prgs.cc",
identity="jcwalker3",
org=INSTALL_ORG,
repository=INSTALL_REPO,
role_kind="author",
source="test-2",
canonical_repository_root=self.install_root,
)
bound = session_ctx.get_session_context()
self.assertEqual(bound["canonical_repository_root"], self.target_root)
self.assertEqual(bound["repository"], TARGET_REPO)
def test_request_supplied_worktree_cannot_replace_the_root(self):
with self._bind("author", self.target_root):
ctx = srv._resolve_namespace_mutation_context(self.install_root)
# The workspace argument may be demoted/inspected, but the canonical
# repository root stays the configured target.
self.assertEqual(ctx["canonical_repo_root"], self.target_root)
def test_request_supplied_coordinates_cannot_replace_the_root(self):
with self._bind("author", self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
self.assertEqual(srv._canonical_local_git_root(), self.target_root)
# ===========================================================================
# 7. Parity dimensions stay separately labelled (#739 F3 preserved)
# ===========================================================================
class TestParityDimensionSeparation(_CrossRepoHarness):
"""Server-implementation parity stays anchored to the install checkout."""
def test_server_dimension_is_installation_not_target(self):
import master_parity_gate
with self._bind("author", self.target_root):
head = master_parity_gate.read_git_head(srv.PROJECT_ROOT)
self.assertEqual(head, _git(self.install_root, "rev-parse", "HEAD"))
self.assertNotEqual(head, _git(self.target_root, "rev-parse", "HEAD"))
def test_target_dimension_reads_the_target_checkout(self):
with self._bind("author", self.target_root):
self.assertEqual(
_git(srv._canonical_local_git_root(), "rev-parse", "HEAD"),
_git(self.target_root, "rev-parse", "HEAD"),
)
# ===========================================================================
# 8. Configuration loaders validate the binding consistently (AC13)
# ===========================================================================
class TestConfigurationLoaderValidation(unittest.TestCase):
"""Every supported loader treats canonical_repository_root identically."""
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.tmp = self._dir.name
def tearDown(self):
self._dir.cleanup()
def _write(self, data: dict) -> str:
path = os.path.join(self.tmp, "profiles.json")
with open(path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(data))
return path
# --- v1 ---------------------------------------------------------------
def _v1(self, root):
return {
"version": 1,
"profiles": {
"prgs-author": {
"base_url": "https://gitea.prgs.cc",
"auth": {"type": "env", "name": "GITEA_TOKEN"},
"canonical_repository_root": root,
}
},
}
def test_v1_rejects_relative_canonical_root(self):
path = self._write(self._v1("relative/path"))
with self.assertRaises(gitea_config.ConfigError) as ctx:
gitea_config.load_config(path)
self.assertIn("absolute", str(ctx.exception))
def test_v1_rejects_blank_canonical_root(self):
path = self._write(self._v1(" "))
with self.assertRaises(gitea_config.ConfigError):
gitea_config.load_config(path)
def test_v1_accepts_absolute_canonical_root(self):
path = self._write(self._v1("/abs/target"))
loaded = gitea_config.load_config(path)
self.assertEqual(
loaded["profiles"]["prgs-author"]["canonical_repository_root"],
"/abs/target",
)
def test_v1_without_the_field_is_unchanged(self):
data = self._v1("/abs/target")
del data["profiles"]["prgs-author"]["canonical_repository_root"]
loaded = gitea_config.load_config(self._write(data))
self.assertNotIn("canonical_repository_root", loaded["profiles"]["prgs-author"])
# --- v2 environments --------------------------------------------------
def _v2_env(self, root):
ident = {
"auth": {"type": "env", "name": "GITEA_TOKEN"},
"base_url": "https://gitea.prgs.cc",
"username": "jcwalker3",
"allowed_operations": ["gitea.read"],
}
if root is not None:
ident["canonical_repository_root"] = root
return {
"version": 2,
"environments": {
"prgs": {"services": {"gitea": {"identities": {"author": ident}}}}
},
}
def test_v2_environments_propagates_canonical_root(self):
"""Regression: the field was silently dropped during flattening."""
loaded = gitea_config.load_config(self._write(self._v2_env("/abs/target")))
profile = loaded["profiles"]["prgs.gitea.author"]
self.assertEqual(profile["canonical_repository_root"], "/abs/target")
def test_v2_environments_rejects_relative_canonical_root(self):
with self.assertRaises(gitea_config.ConfigError) as ctx:
gitea_config.load_config(self._write(self._v2_env("relative/path")))
self.assertIn("absolute", str(ctx.exception))
def test_v2_environments_without_the_field_is_unchanged(self):
loaded = gitea_config.load_config(self._write(self._v2_env(None)))
self.assertNotIn(
"canonical_repository_root", loaded["profiles"]["prgs.gitea.author"]
)
# --- v2 contexts (already validated; guard against regression) ---------
def test_v2_contexts_still_rejects_relative_canonical_root(self):
data = {
"version": 2,
"contexts": {
"prgs": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "prgs",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN"},
"allowed_operations": ["gitea.read"],
"canonical_repository_root": "relative/path",
}
},
}
with self.assertRaises(gitea_config.ConfigError):
gitea_config.load_config(self._write(data))
# ===========================================================================
# 9. Role-by-operation matrix
# ===========================================================================
class TestRoleOperationMatrix(_CrossRepoHarness):
"""Each role, each cross-repository condition, one assertion per cell."""
def test_correct_target_resolves_for_every_role(self):
for role in ROLES:
with self.subTest(role=role, case="correct-target"):
with self._bind(role, self.target_root):
_h, org, repo = srv._resolve("prgs", None, None, None)
self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
def test_wrong_repository_is_rejected_for_every_role(self):
for role in ROLES:
with self.subTest(role=role, case="wrong-repo"):
with self._bind(role, self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
def test_explicit_matching_coordinates_pass_for_every_role(self):
for role in ROLES:
with self.subTest(role=role, case="explicit-match"):
with self._bind(role, self.target_root):
_h, org, repo = srv._resolve("prgs", None, TARGET_ORG, TARGET_REPO)
self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
def test_explicit_mismatch_fails_closed_for_every_role(self):
for role in ROLES:
with self.subTest(role=role, case="explicit-mismatch"):
with self._bind(role, self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, INSTALL_ORG, THIRD_REPO)
def test_missing_canonical_root_preserves_single_repo_default(self):
for role in ROLES:
with self.subTest(role=role, case="missing-root"):
with self._bind(role, None):
self.assertEqual(srv._canonical_local_git_root(), self.install_root)
def test_invalid_canonical_root_never_becomes_install_root(self):
for role in ROLES:
with self.subTest(role=role, case="invalid-root"):
with self._bind(role, self.not_a_repo):
self.assertNotEqual(
srv._canonical_local_git_root(), self.install_root
)
def test_wrong_worktree_rejected_for_every_role(self):
import author_mutation_worktree as amw
foreign = os.path.join(self.install_root, "branches", "foreign")
os.makedirs(foreign, exist_ok=True)
for role in ROLES:
with self.subTest(role=role, case="wrong-worktree"):
with self._bind(role, self.target_root):
membership = amw.assess_workspace_repo_membership(
workspace_path=foreign,
canonical_repo_root=srv._canonical_local_git_root(),
)
self.assertTrue(membership["block"])
def test_request_override_attempt_rejected_for_every_role(self):
for role in ROLES:
with self.subTest(role=role, case="request-override"):
with self._bind(role, self.target_root):
with self.assertRaises(RuntimeError):
srv._resolve("prgs", None, "Attacker", "evil-repo")
self.assertEqual(srv._canonical_local_git_root(), self.target_root)
if __name__ == "__main__":
unittest.main()