genichin/gitea-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[verified] Enforce delegation adoption at runtime consumers

Browse Source
This commit is contained in:
gitea-spec-agent
2026-10-11 07:11:40 +00:00
parent 9367d5a3a1
commit 6cb2317667
19 changed files with 1112 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.py
+40 -4
View File
@@ -18,6 +18,23 @@ IDENT = re.compile(r"^[^\s/\\\x00<>`=@]+$")
CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f\u2028\u2029]")
MARKUP = re.compile(r"[`<>\[\]{}]|(?:^|\s)[#*!|]|[a-z][a-z0-9+.-]*://|www\.", re.I)
def _require_adoption(declaration, revision, stage):
if declaration is None:
_fail("delegation adoption declaration required")
try:
repo_root = Path(__file__).resolve().parents[6]
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from harness.lib.delegation_adoption import AdoptionError, require_adoption
except Exception:
_fail("delegation adoption unavailable")
try:
return require_adoption(declaration, expected_source={"kind": "git_tree", "revision": revision}, expected_stage=stage)
except AdoptionError as exc:
_fail(str(exc))
except Exception:
_fail("delegation adoption unavailable")
class ValidationError(ValueError):
pass
@@ -179,7 +196,9 @@ def _marker(r, digest):
return ("<!-- gitea-agent:implementation-spec-review/v1 owner=%s repository=%s issue=%d pr=%d head=%s contract-sha256=%s result-sha256=%s correlation=%s -->" %
(t["owner"],t["repository"],t["issue"],t["pull"],t["head"],r["contract_sha256"],digest,corr))
def render_review(result, allowed_paths, sensitive_values=()):
def render_review(result, allowed_paths, sensitive_values=(), *, declaration=None):
if type(result) is dict and type(result.get("target")) is dict:
_require_adoption(declaration, result["target"].get("head"), "spec-review")
r=validate_result(result, allowed_paths, sensitive_values); t=r["target"]
lines=["## 구현 명세 적합성 검토", "", f"- Source Issue: #{t['issue']}", f"- Pull Request: #{t['pull']}",
f"- Reviewed head: `{t['head']}`", f"- Contract SHA-256: `{r['contract_sha256']}`", f"- Verdict: `{r['verdict']}`", "",
@@ -233,7 +252,7 @@ def parse_review(body, allowed_paths, sensitive_values=()):
except (IndexError, ValueError): _fail("invalid review body")
r={"schema":SCHEMA,"target":{"owner":m.group(1),"repository":m.group(2),"issue":issue,"pull":pull,"head":head},"contract_sha256":contract,"verdict":verdict,"criteria":criteria,"scope":scope,"conclusion":conclusion}
validate_result(r, allowed_paths, sensitive_values)
canonical=render_review(r, allowed_paths, sensitive_values)
canonical=render_review(r, allowed_paths, sensitive_values, declaration={"mode": "none"})
if canonical != body or _sha(canonical[:canonical.rfind("<!--") - 1].encode()) != m.group(7): _fail("review digest mismatch")
if _marker(r,m.group(7)) != marker: _fail("review marker mismatch")
return r
@@ -547,7 +566,7 @@ def verify_publication(before, after, exact_review, result, allowed_paths, revie
if not _same(er, exact_review):
_fail("exact review readback mismatch")
expected = {"approve": "APPROVED", "request_changes": "REQUEST_CHANGES", "blocked_spec": "COMMENT"}[result["verdict"]]
if er.get("body") != render_review(result, allowed_paths) or er.get("commit_id") != target["head"] or er.get("state") != expected:
if er.get("body") != render_review(result, allowed_paths, declaration={"mode": "none"}) or er.get("commit_id") != target["head"] or er.get("state") != expected:
_fail("review binding mismatch")
if _user_id(er) != reviewer_id or er.get("pull_request_url") != url:
_fail("review identity mismatch")
@@ -634,6 +653,10 @@ def main(argv=None):
q=sub.add_parser(name); q.add_argument("--result",required=True); q.add_argument("--allowed-paths",required=True); return q
common("validate-result"); common("render")
q=sub.add_parser("parse"); q.add_argument("--body",required=True); q.add_argument("--allowed-paths",required=True)
render_parser = sub.choices["render"]
adoption = render_parser.add_mutually_exclusive_group(required=True)
adoption.add_argument("--no-delegated-evidence", action="store_true")
adoption.add_argument("--delegation-adoption")
q=sub.add_parser("check-cache"); q.add_argument("--contract-path",required=True); q.add_argument("--owner",required=True); q.add_argument("--repository",required=True); q.add_argument("--issue",required=True,type=int)
q=sub.add_parser("check-pages"); q.add_argument("--pages",required=True); q.add_argument("--key",required=True)
q.add_argument("--page-size", type=int, default=50); q.add_argument("--max-pages", type=int, default=10000)
@@ -644,7 +667,20 @@ def main(argv=None):
q.add_argument("--expected-pr-url", required=True)
try:
a=p.parse_args(argv)
if a.cmd in ("validate-result","render"): r=_json_file(a.result); paths=_json_file(a.allowed_paths); validate_result(r,paths); out=r if a.cmd=="validate-result" else render_review(r,paths); print(json.dumps(out,ensure_ascii=False,sort_keys=True) if a.cmd=="validate-result" else out,end="" if a.cmd=="render" else "\n")
if a.cmd in ("validate-result","render"):
r=_json_file(a.result); paths=_json_file(a.allowed_paths); validate_result(r,paths)
if a.cmd == "validate-result": out=r
else:
if a.no_delegated_evidence:
declaration={"mode":"none"}
else:
repo_root=Path(__file__).resolve().parents[6]
if str(repo_root) not in sys.path:
sys.path.insert(0,str(repo_root))
from harness.lib.delegation_evidence import _json_file as load_declaration
declaration=load_declaration(a.delegation_adoption)
out=render_review(r,paths,declaration=declaration)
print(json.dumps(out,ensure_ascii=False,sort_keys=True) if a.cmd=="validate-result" else out,end="" if a.cmd=="render" else "\n")
elif a.cmd=="parse": print(json.dumps(parse_review(_read(a.body),_json_file(a.allowed_paths)),ensure_ascii=False,sort_keys=True))
elif a.cmd=="check-cache": print(json.dumps(validate_cache_paths(a.contract_path,a.owner,a.repository,a.issue),sort_keys=True))
elif a.cmd=="check-pages":