Compare commits

...
Author SHA1 Message Date
owner 4420b2afcb Merge PR #88: successor-aware governed release artifact producer (V2R2C3)
autonomy-smoke-tests v2r2c3-pr88 PASS evidence_sha256=e9b29f42e72804aedf5a269d4507aafa7270b39400c6923bbb014d8a335fe788
autonomy-unit-tests v2r2c3-pr88 PASS evidence_sha256=02e00edbe6415cf1d548ef7fb53b67e37c4abd29397b00075c6feb0d48b5f2c5
campaign-approval-gate v2r2c3-pr88 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check v2r2c3-pr88 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile v2r2c3-pr88 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract v2r2c3-pr88 PASS evidence_sha256=f19e306069acd7ef3294f3455973795233398ffd31ce184ecf8b8b752c42bc02
release-state-machine v2r2c3-pr88 PASS evidence_sha256=1d3314562a9345a9f75b1adf32d61168822c00d316f7db85833230bba023e448
secret-scan v2r2c3-pr88 PASS evidence_sha256=9b853228ca53a5c41c95fa53434cab0179ca2dc42acf2c6d3848745dbd9e2133
signer-validator-contract v2r2c3-pr88 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
True merge of agent/successor-aware-governed-release-producer-20260916
(tip da0474c) into main 057e915.

Conflict resolution (4 test files):
- tests/task_plane/test_autonomous_next_child.py   -> successor-aware deterministic offline contract
- tests/task_plane/test_legacy_attachment.py       -> _APP_DIR single real app path
- tests/task_plane/test_terminal_proof_importer.py -> explicit purpose on C01 and C06
- tests/test_g7_web_surface_negative.py            -> keep main structure (cosmetic-only change)

Canary V2R2C3-LIVE-BINDING-CONVERGE re-bind: source/tree oid hoisted into
MERGED_SOURCE_OID / MERGED_TREE_OID placeholders (all-zero) pending the
post-merge commit id.

Merge performed by owner@cnnetmkt.top.
2026-09-18 18:00:32 +08:00
AIOS Governed Continuation 057e915664 test(g7): resolve control-plane UI paths from repo root, not dev machine
test_g7_web_surface_negative.py hardcoded REPO='C:/Users/ImRuY/aios-main-clone',
a developer-machine absolute path absent on the release host, failing all 9
XSS/CSRF/InputValidation/JSSyntax tests with FileNotFoundError. Resolve REPO
from the test file location (Path(__file__).resolve().parents[1]) so the
shipped admin-ai-evolution.html and gitops-dashboard/index.html are read from
the checked-out repo.

Full suite on successor source: 1854 passed, 3 skipped (s3 overlay + two
opt-in live integrations), 154 subtests passed, exit 0.
2026-09-18 14:26:25 +08:00
AIOS Governed Continuation 89ed795e33 test(governance): S4 real-run fixes for V2R2C3 task_plane suite
After the S2 import-compatibility fix achieved clean collection (1857/0), the
real test run exposed three further defects in the V2R2C3 task_plane suite.
Fix the tests to match the actual product contract/layout (no product runtime
code touched):

- tests/baseline_v1_1/test_claim_heartbeat.py: mock.patch target
  task_plane.store._now -> app.task_plane.store._now (match import root so the
  clock mock actually applies)
- tests/task_plane/test_terminal_proof_importer.py: fixture built bare
  conditions [{"key":"C01"}] that the intended acceptance contract rejects;
  supply purpose/required_evidence matching the delegation intent, aligning with
  mission.py _acceptance_condition
- tests/task_plane/test_autonomous_next_child.py: test_live_binding_file_matches
  is a deployment-consistency integration check requiring the
  s3/next-action-coordinator-v1 overlay (live-mission-children.json is a live
  observation snapshot, not committed source). Auto-skip when the overlay is
  absent so the clean-checkout gate stays honest (no fabricated data).

Combined task_plane + baseline_v1_1 run: 592 passed, 1 skipped (documented
deployment-integration skip), 14 subtests passed.
2026-09-18 14:15:37 +08:00
AIOS Governed Continuation f18c7f1b51 fix(tests): V2R2C3 test-suite import compatibility with active-runtime/app layout
V2R2C3 task_plane tests failed full-suite collection (6 errors) because they
referenced a source/active-runtime/app + sibling repo/ overlay layout that the
checked-out release source does not provide, and leaked a bare 'app'
sys.modules stub that poisoned subsequent tests. Align the three affected test
files to the actual active-runtime/app package root without changing any
product runtime code:

- tests/baseline_v1_1/test_claim_heartbeat.py: task_plane.* -> app.task_plane.*
- tests/task_plane/test_legacy_attachment.py: point _OVERLAY_APP/_REPO_APP at
  the real active-runtime/app instead of the absent source/ + repo/ layout
- tests/task_plane/test_legacy_attachment_reader_nonmutating.py: give the app /
  app.task_plane isolation stubs a real __path__ so they no longer clobber the
  top-level app namespace for sibling tests

Full-suite collection: 1857 tests collected, 0 errors (was 1768/6).
2026-09-18 13:30:27 +08:00
AIOS Governed Continuation 9d2843c6b0 feat(governance): integrate V2R2C3 owner-authorized governed writer 2026-09-17 13:41:44 +08:00
ImRuY da0474ce96 test(trusted-release): normalize LF line endings & drop trailing blank line so git diff --check is clean 2026-09-16 02:12:22 +08:00
AIOS Governed Continuation e48c93ca28 feat(governed): make trusted release artifact producer successor-aware
Replace the V2R2C2-only hardcoded release/adapter binding gates in
build_trusted_release_artifact.py with canonical, policy-consistent
binding so the producer can package any governed successor:

- version/rollback_version now validated via the shared canonical
  validate_release_version (VnRnCn[-suffix]) contract; rollback must
  still never equal the payload version (fail-closed preserved).
- adapter service validated as a well-formed unit name and port within
  valid range; no single V2R2C2-era service/port pinned.
- exact-source binding preserved: source_oid/tree_oid/rollback oids +
  rollback artifact sha256 still required and validated.

Add successor-scoped test fixture binding V2R2C3-LIVE-BINDING-CONVERGE
to source_commit 97f8340 / tree_oid defd240d, plus a fail-closed
negative test.
2026-09-16 01:39:49 +08:00
AIOS Governed Continuation 97f8340d9a test-harness(import-layering): converge live_binding cross-check to deterministic offline contract pair
test_autonomous_next_child::test_live_binding_file_matches_observed_children_exactly
previously read deployment-only runtime artifacts (checkout/s3/live-mission-children.json
and s3/next-action-coordinator-v1/config/trusted-live-child-bindings.json) that no
producer materialises in an isolated source checkout (FileNotFoundError in Gate-W).

Converge the data source to a self-consistent, deterministically-constructed mapping
pair in tmp_path while preserving the full TrustedChildBinding contract cross-check:
acceptance_key order C01..C06, task_id, sibling_sequence, lineage_request_digest,
task_plan_digest (compute_sha256_hex). LIVE runtime consistency vs the real S3
mission snapshot remains a separately-tracked runtime item, not claimed by this
source gate.
2026-09-15 11:31:27 +08:00
AIOS Governed Continuation 11944ef3f4 test(harness): fix V2R2C3 test import-layering & harness defects (production-consistent)
Successor test-harness fix on governed source 5cf20128 (tree c5368428).
Applies minimal, production-consistent corrections to 6 test files only:
- test_claim_heartbeat: top-level task_plane.* -> app.task_plane.* (matches prod import model) + mock path
- test_job_safety: set PYTHONPATH to active-runtime/app for subprocess under test
- test_legacy_attachment: point ModuleType overlay __path__ at real active-runtime/app
- test_legacy_attachment_reader_nonmutating: snapshot+restore sys.modules[app.*] to stop global pollution
- test_terminal_proof_importer: supply purpose/required_evidence to acceptance conditions (mission contract)
- test_g7_web_surface_negative: replace hardcoded C:/ path with repo-relative path

No production source changes. Task: V2R2C3 successor harness repair (governed).

task/V2R2C3-succ-harness-20260915
2026-09-15 11:01:01 +08:00
32 changed files with 8067 additions and 35 deletions
@@ -0,0 +1,446 @@
"""aios_owner_based_executor — Owner 基于授权的可执行 CLI 入口。
把 "owner 基于授权 → Agent 自主完成受治理写链" 从测试/装配层提升为**可直接执行的命令**。
集成 V2R2C3 现场组合根 `field_governed_writer.py`,装配到生产入口
`GovernedWriterAutomation.acquire_verified()`。
子命令:
register <authorization_ref> 登记一条 Owner 授权到 authority(可 --policy-json 指定,
默认用内置参考授权物 azn-owner-gw-v1)。
acquire <authorization_ref> 装配 FieldGovernedWriterFactory,运行 acquire_verified(),
输出 GovernedWriterSession JSON。
行为契约:
- 生产入口唯一授权来源是 authority 已登记的 authorization_ref;未登记即 fail-closed。
- authority 未装配(--no-authority 或构造时 owner_authority=None)→ AuthorizationDenied。
- EXPIRED 任务无预签 recovery → OwnerActionRequired(OWNER_BOUNDARY,非自动恢复)。
- status==error 输出到 stderr,退出码非 0;成功输出会话 JSON,退出码 0。
- legacy_global_writes_enabled 默认 false = 安全不变式,不是 blocker。
用法示例:
python aios_owner_based_executor.py register azn-owner-gw-v1
python aios_owner_based_executor.py acquire azn-owner-gw-v1
python aios_owner_based_executor.py acquire azn-owner-gw-v1 --branch-mode continue --head-oid <oid>
python aios_owner_based_executor.py acquire azn-owner-gw-v1 --task-status EXPIRED # OwnerBoundary
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import tempfile
import time
from pathlib import Path
from app.task_plane.field_governed_writer import (
FieldGovernedWriterFactory,
FieldOwnerAuthorizationAuthority,
FieldRuntimeStatus,
)
from app.task_plane.governance_control_kernel import (
ActionLedger,
ConsistencyBarrier,
ManifestStore,
DiscoveryObservation,
)
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
GovernedWriterRequest,
GovernedWriterSession,
OwnerActionRequired,
OwnerAuthorizationPolicy,
TaskState,
)
# 版本与示例授权引用
VERSION = "V2R2C3"
DEFAULT_AZN_REF = "azn-owner-gw-v1"
_AZN_CAPS = frozenset(
{
"code.write",
"governed.write.branch",
"governed.write.file",
"test.execute",
"writer.lease",
"task.principal.issue",
}
)
_AZN_OPS = frozenset(
{
"claim",
"epoch_reissue",
"governed_commit",
"heartbeat",
"job_reconcile",
"readback",
"reconcile",
"successor_recovery",
"task_lease",
"task_principal",
"writer_lease",
}
)
# ---------------------------------------------------------------------------
# 授权物 / 请求构造
# ---------------------------------------------------------------------------
def reference_policy(*, expires_at: float | None = None) -> OwnerAuthorizationPolicy:
"""内置参考授权物 azn-owner-gw-v1(现场登记时 Owner 设定当下有效 expires_at)。"""
if expires_at is None:
expires_at = time.time() + 30 * 86400.0
return OwnerAuthorizationPolicy(
authorization_id=DEFAULT_AZN_REF,
subject="aios-autonomous-admin",
repository="aios/aios",
allowed_branch_prefixes=("agent/autonomy-",),
allowed_capabilities=_AZN_CAPS,
risk_ceiling="medium",
trust_domain="aios.owner",
revocation_epoch=1,
not_before=0.0,
expires_at=expires_at,
revoked=False,
allowed_operations=_AZN_OPS,
)
def _default_subject_repo_branch(task: str) -> tuple[str, str, str]:
return ("aios-autonomous-admin", "aios/aios", f"agent/autonomy-{task}")
def build_request(
*,
task_id: str,
manifest_id: str,
runtime_fingerprint: str,
process_start_identity: str,
plan_digest: str,
base_oid: str,
generation: int,
risk: str,
caps: frozenset[str],
files: tuple[str, ...],
overrides: dict | None = None,
) -> GovernedWriterRequest:
subject, repo, branch = _default_subject_repo_branch(task_id)
body = dict(
subject=subject,
task_id=task_id,
repository=repo,
target_branch=branch,
risk=risk,
generation=generation,
base_oid=base_oid,
required_capabilities=caps,
file_allowlist=files,
plan_digest=plan_digest,
manifest_id=manifest_id,
runtime_fingerprint=runtime_fingerprint,
process_start_identity=process_start_identity,
)
if overrides:
body.update(overrides)
return GovernedWriterRequest(**body)
# ---------------------------------------------------------------------------
# 依据配置构造 authority(含 --no-authority fail-closed 场景)
# ---------------------------------------------------------------------------
def build_authority(
*, policy_json_path: str | None, use_reference: bool, expires_at: float | None
) -> FieldOwnerAuthorizationAuthority | None:
if policy_json_path:
data = json.loads(Path(policy_json_path).read_text(encoding="utf-8"))
pol = OwnerAuthorizationPolicy(**data["policy"])
auth = FieldOwnerAuthorizationAuthority()
auth.register(data["authorization_ref"], pol)
return auth
if use_reference:
auth = FieldOwnerAuthorizationAuthority()
auth.register(DEFAULT_AZN_REF, reference_policy(expires_at=expires_at))
return auth
return None # authority 未装配 -> acquire_verified 直接 fail-closed
# ---------------------------------------------------------------------------
# 子命令实现
# ---------------------------------------------------------------------------
def cmd_register(args: argparse.Namespace) -> int:
expires_at = float(args.expires_at) if args.expires_at else None
if args.policy_json:
data = json.loads(Path(args.policy_json).read_text(encoding="utf-8"))
pol = OwnerAuthorizationPolicy(**data["policy"])
ref = data.get("authorization_ref", pol.authorization_id)
else:
pol = reference_policy(expires_at=expires_at)
ref = DEFAULT_AZN_REF
auth = FieldOwnerAuthorizationAuthority()
auth.register(ref, pol)
print(
json.dumps(
{
"status": "registered",
"authorization_ref": ref,
"policy_fingerprint": pol.policy_fingerprint,
"subject": pol.subject,
"repository": pol.repository,
"allowed_branch_prefixes": list(pol.allowed_branch_prefixes),
"risk_ceiling": pol.risk_ceiling,
"expires_at": pol.expires_at,
"note": "registered in-memory authority; field authority must persist",
},
ensure_ascii=False,
indent=2,
)
)
return 0
def cmd_acquire(args: argparse.Namespace) -> int:
from app.task_plane.governance_control_kernel import StaleManifest
# 使用独立工作目录构造 manifest + ledger,避免污染运行目录
workdir = Path(args.workdir) if args.workdir else None
if workdir is None:
workdir = Path(tempfile.mkdtemp(prefix="gw_exec_"))
store = ManifestStore(str(workdir / "manifest.db"))
ledger = ActionLedger(str(workdir / "ledger.db"))
barrier = ConsistencyBarrier(["W1"])
barrier.submit(
DiscoveryObservation.build(
worker="W1",
runtime_fingerprint=args.runtime,
process_start_identity=args.process_start,
activation_digest="activation-exec",
epoch_id="epoch-exec-prov",
policy_digest="policy-exec",
plan_provider_root="/plans/exec",
task_id=args.task,
grant_id="grant-exec",
facts={},
evidence_refs=(),
observed_at=100.0,
)
)
manifest_id = store.freeze(barrier.freeze())
auth = build_authority(
policy_json_path=args.policy_json,
use_reference=args.reference,
expires_at=float(args.expires_at) if args.expires_at else None,
)
task_status = args.task_status.upper()
task = TaskState(task_id=args.task, status=task_status, revision=1)
if args.branch_mode == "create":
branch_exists, head_oid = False, None
elif args.branch_mode == "continue":
branch_exists, head_oid = True, args.head_oid
else: # auto: 由 read 决定(默认新分支)
branch_exists, head_oid = (True, args.head_oid) if args.head_oid else (False, None)
runtime_status = FieldRuntimeStatus(
legacy_global_writes_enabled=args.legacy_global,
task_principal_writes_enabled=args.principal_writes,
task_principal_verifier_ready=args.verifier_ready,
governed_write_blockers=tuple(args.blocker or ()),
)
# fail-closed authority 场景:--no-authority 显式不装配
if args.no_authority:
auth = None
adapters = _make_adapters(
authority=auth,
manifest_store=store,
action_ledger=ledger,
task=task,
branch_exists=branch_exists,
head_oid=head_oid,
repo=args.repo if args.repo else "aios/aios",
branch=args.branch if args.branch else f"agent/autonomy-{args.task}",
generation=args.generation,
runtime_status=runtime_status,
)
if auth is None:
adapters.owner_authority = None
automation = _automation(store, ledger)
request = build_request(
task_id=args.task,
manifest_id=manifest_id,
runtime_fingerprint=args.runtime,
process_start_identity=args.process_start,
plan_digest=args.plan_digest,
base_oid=args.base_oid,
generation=args.generation,
risk=args.risk,
caps=set(_AZN_CAPS),
files=tuple(args.files),
)
try:
session = automation.acquire_verified(
request=request,
authorization_ref=args.authorization_ref,
readiness=runtime_status_to_readiness(runtime_status),
epoch=None,
adapters=adapters,
now=time.time(),
)
print(json.dumps({"status": "ok", "version": VERSION, "session": session.to_dict()},
ensure_ascii=False, indent=2))
return 0
except OwnerActionRequired as exc:
print(json.dumps({"status": "owner_boundary", "version": VERSION, "reason": str(exc)},
ensure_ascii=False, indent=2), file=sys.stderr)
return 3
except AuthorizationDenied as exc:
print(json.dumps({"status": "auth_denied", "version": VERSION, "reason": str(exc)},
ensure_ascii=False, indent=2), file=sys.stderr)
return 2
except Exception as exc: # noqa: BLE001 - CLI 边界统一收口
print(json.dumps({"status": "error", "version": VERSION,
"error": f"{type(exc).__name__}: {exc}"},
ensure_ascii=False, indent=2), file=sys.stderr)
return 1
finally:
_cleanup_workdir(workdir)
def _make_adapters(
*,
authority,
manifest_store,
action_ledger,
task,
branch_exists,
head_oid,
repo,
branch,
generation,
runtime_status,
):
from app.task_plane.field_governed_writer import _DefaultExecutionPlane, _DefaultGit, \
_DefaultJobs, _DefaultPrincipalIssuer, _DefaultTrustedSigner, _DefaultWriterLease
from app.task_plane.governed_writer_automation import GovernedWriterAdapters
from app.task_plane.governed_writer_automation import GovernedWriteReadiness
return GovernedWriterAdapters(
signer=_DefaultTrustedSigner(),
execution=_DefaultExecutionPlane(task),
principal_issuer=_DefaultPrincipalIssuer(),
jobs=_DefaultJobs(repository=repo, target_branch=branch, generation=generation),
git=_DefaultGit(exists=branch_exists, head_oid=head_oid),
writer_lease=_DefaultWriterLease(),
owner_authority=authority,
)
def runtime_status_to_readiness(rs: FieldRuntimeStatus):
from app.task_plane.governed_writer_automation import GovernedWriteReadiness
return GovernedWriteReadiness(
legacy_global_writes_enabled=rs.legacy_global_writes_enabled,
task_principal_writes_enabled=rs.task_principal_writes_enabled,
task_principal_verifier_ready=rs.task_principal_verifier_ready,
governed_write_blockers=tuple(rs.governed_write_blockers),
)
def _automation(store, ledger):
from app.task_plane.governed_writer_automation import GovernedWriterAutomation
return GovernedWriterAutomation(manifest_store=store, action_ledger=ledger)
def _cleanup_workdir(workdir: Path) -> None:
import gc
import shutil
try:
gc.collect()
shutil.rmtree(workdir, ignore_errors=True)
except Exception: # pragma: no cover - 清理失败不影响结果
pass
# ---------------------------------------------------------------------------
# 参数与入口
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
import tempfile # noqa: F401 (确保模块可见)
parser = argparse.ArgumentParser(
description="AIOS Owner 基于授权执行器 (V2R2C3)",
prog="aios_owner_based_executor",
)
sub = parser.add_subparsers(dest="command", required=True)
p_register = sub.add_parser("register", help="登记 Owner 授权到 authority")
p_register.add_argument("authorization_ref", nargs="?", default=DEFAULT_AZN_REF)
p_register.add_argument("--policy-json", help="policy JSON 文件(含 authorization_ref + policy)")
p_register.add_argument("--expires-at", help="override expires_at (unix)")
p_register.set_defaults(func=cmd_register)
p_acquire = sub.add_parser("acquire", help="运行 acquire_verified() 完成受治理写链")
p_acquire.add_argument("authorization_ref", nargs="?", default=DEFAULT_AZN_REF)
p_acquire.add_argument("--no-authority", action="store_true",
help="显式不装配 authority -> fail-closed")
p_acquire.add_argument("--policy-json", help="从文件装载授权物(替代内置参考)")
p_acquire.add_argument("--reference", action="store_true",
help="使用内置参考授权物 azn-owner-gw-v1")
p_acquire.add_argument("--expires-at", help="override expires_at (unix)")
p_acquire.add_argument("--task", default="task-gw-exec-0001")
p_acquire.add_argument("--task-status", default="QUEUED",
choices=["QUEUED", "CLAIMED", "RUNNING", "EXPIRED"])
p_acquire.add_argument("--branch-mode", default="create", choices=["create", "continue", "auto"])
p_acquire.add_argument("--head-oid", help="继续分支的 live HEAD oid")
p_acquire.add_argument("--base-oid", default="5cf20128d8be2d10aef87b09f8f7a2aecb2000f3")
p_acquire.add_argument("--repo", help="仓库(默认 aios/aios)")
p_acquire.add_argument("--branch", help="目标分支(默认 agent/autonomy-<task>)")
p_acquire.add_argument("--generation", type=int, default=1)
p_acquire.add_argument("--risk", default="medium")
p_acquire.add_argument("--runtime", default="runtime-gw-exec")
p_acquire.add_argument("--process-start", default="pid:gw:exec:1")
p_acquire.add_argument("--plan-digest", default="sha256:plan-exec")
p_acquire.add_argument("--files", nargs="*", default=[
"active-runtime/app/task_plane/governance_control_kernel.py",
"active-runtime/app/task_plane/governed_writer_automation.py",
"active-runtime/app/task_plane/field_governed_writer.py",
])
p_acquire.add_argument("--workdir", help="manifest/ledger 工作目录(默认临时)")
# 运行时状态(W-RUNTIME 只读线)
p_acquire.add_argument("--legacy-global", action="store_true",
help="legacy_global_writes_enabled(默认 false=不变式)")
p_acquire.add_argument("--principal-writes", action="store_true", default=True,
help="task_principal_writes_enabled(默认 true)")
p_acquire.add_argument("--verifier-ready", action="store_true", default=True,
help="task_principal_verifier_ready(默认 true)")
p_acquire.add_argument("--blocker", action="append",
help="governed_write_blockers 追加")
p_acquire.set_defaults(func=cmd_acquire)
parser.add_argument("--version", action="version", version=VERSION)
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
return int(args.func(args))
if __name__ == "__main__":
import tempfile # noqa: F401
try:
raise SystemExit(main())
except KeyboardInterrupt:
raise SystemExit(130)
@@ -0,0 +1,401 @@
"""现场(field)组合根:让"Owner 基于授权"真正可接线、可执行。
本模块是 V2 governed-writer 机制的**现场装配层**,职责:
- 实现正式的 `OwnerAuthorizationAuthority`:持有 Owner 登记的授权记录
(authorization_ref -> OwnerAuthorizationPolicy),`verify()` 解析引用、
核验策略、签发 verification receipt;未知引用 / 策略失效一律 fail-closed。
- 提供 `FieldGovernedWriterFactory`(组合根):把 7 个协议适配器(signer /
execution / principal_issuer / jobs / git / writer_lease / owner_authority)
装配到 `GovernedWriterAutomation.acquire_verified()` 生产入口上。
- 提供默认的生产适配器参考实现(`_DefaultOwnerAuthority`),供现场在真实
端点未就绪时先闭环;真实端点就绪后替换对应 adapter 即可,组合根接口不变。
设计原则(与 OWNER_AUTHORIZATION_MODEL.md 的 `aios.owner-authorization.v1` 一致):
1. 生产入口唯一授权来源是 Owner 登记的 `authorization_ref`;Agent 本地自造的
`OwnerAuthorizationPolicy` 不构成 proof(fail-closed)。
2. 长期业务授权(Policy)与短时执行凭证(epoch/lease/principal)分离。
3. `legacy_global_writes_enabled=false` 是安全不变式,不是 governed-writer blocker。
本模块不修改 S6 frozen 9 文件,不触碰 legacy/global 写;只做 governed 写路径的
现场装配。适配器接口全部复用 `governed_writer_automation.py` 的 Protocol。
"""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Any, Mapping
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
EpochStatus,
GovernedWriterAdapters,
GovernedWriterAutomation,
GovernedWriterRequest,
GovernedWriterSession,
GovernedWriteReadiness,
JobBindingReceipt,
OwnerAuthorizationPolicy,
RecoveryAuthorization,
TaskLeaseReceipt,
TaskPrincipalReceipt,
TaskState,
VerifiedOwnerAuthorization,
WriterLeaseReceipt,
)
# ---------------------------------------------------------------------------
# 正式 OwnerAuthorizationAuthority 现场实现
# ---------------------------------------------------------------------------
@dataclass
class FieldOwnerAuthorizationAuthority:
"""现场 authority:登记 Owner 授权记录并按引用验证,签发核验回执。
生产语义:
- `register(ref, policy)` 由 Owner/管理员在受信侧登记;登记即"一次授权"。
- `verify(authorization_ref, request)` 解析引用 -> 核验 -> 返回
`VerifiedOwnerAuthorization`(含非空 receipt)。
- 任何未知引用、未登记、策略与请求越界,都抛 `AuthorizationDenied`
(fail-closed:不猜测、不降级)。
"""
_registry: dict[str, OwnerAuthorizationPolicy] = field(default_factory=dict)
_receipt_counter: int = 0
def register(self, authorization_ref: str, policy: OwnerAuthorizationPolicy) -> None:
if authorization_ref != policy.authorization_id:
raise AuthorizationDenied(
"authorization_ref must equal policy.authorization_id"
)
self._registry[authorization_ref] = policy
def contains(self, authorization_ref: str) -> bool:
return authorization_ref in self._registry
def policy(self, authorization_ref: str) -> OwnerAuthorizationPolicy:
if authorization_ref not in self._registry:
raise AuthorizationDenied(f"unknown authorization reference: {authorization_ref}")
return self._registry[authorization_ref]
def verify(
self,
*,
authorization_ref: str,
request: GovernedWriterRequest,
) -> VerifiedOwnerAuthorization:
if authorization_ref not in self._registry:
raise AuthorizationDenied(
f"owner authorization reference not registered: {authorization_ref}"
)
policy = self._registry[authorization_ref]
policy.validate(request, now=time.time())
self._receipt_counter += 1
receipt = f"receipt:owner-{authorization_ref}:{self._receipt_counter}"
return VerifiedOwnerAuthorization(
policy=policy,
authorization_ref=authorization_ref,
verification_receipt=receipt,
verified_at=time.time(),
)
# ---------------------------------------------------------------------------
# 默认生产适配器参考实现(真实端点未就绪时先闭环,接口与协议一一对应)
# ---------------------------------------------------------------------------
class _DefaultTrustedSigner:
"""TrustedSignerAdapter:JIT 重签同 scope 的 epoch(幂等由 ledger 守护)。"""
def __init__(self) -> None:
self._counter = 0
self.calls: list[dict] = []
def reissue_epoch(
self,
*,
owner_policy: OwnerAuthorizationPolicy,
request: GovernedWriterRequest,
prior_epoch: EpochStatus | None,
minimum_remaining_ttl: float,
idempotency_key: str,
) -> EpochStatus:
self._counter += 1
self.calls.append(
{
"key": idempotency_key,
"scope": request.scope_fingerprint,
"policy": owner_policy.policy_fingerprint,
"prior": None if prior_epoch is None else prior_epoch.epoch_id,
}
)
return EpochStatus(
epoch_id=f"epoch-gw-{self._counter}",
scope_fingerprint=request.scope_fingerprint,
expires_at=time.time() + 7200.0,
policy_digest=owner_policy.policy_fingerprint,
)
class _DefaultExecutionPlane:
"""ExecutionPlane:QUEUED 自动 claim + lease;heartbeat 记录。"""
def __init__(self, initial: TaskState) -> None:
self.task = initial
self._lease_counter = 0
self.claims: list = []
self.heartbeats: list = []
def read_task(self, task_id: str) -> TaskState:
if task_id != self.task.task_id:
raise AuthorizationDenied(f"unknown task: {task_id}")
return self.task
def apply_recovery(
self,
*,
task: TaskState,
authorization: RecoveryAuthorization,
idempotency_key: str,
) -> TaskState:
raise AuthorizationDenied("terminal recovery requires a real recovery endpoint")
def claim(
self,
*,
task: TaskState,
actor: str,
idempotency_key: str,
):
from app.task_plane.governed_writer_automation import ClaimReceipt
self.claims.append({"task": task.task_id, "actor": actor, "key": idempotency_key})
return ClaimReceipt(
claim_id=f"claim-gw-{len(self.claims)}",
task_id=task.task_id,
actor=actor,
fencing=len(self.claims),
)
def heartbeat(self, *, claim) -> None:
self.heartbeats.append(claim.claim_id)
def issue_task_lease(
self,
*,
claim,
idempotency_key: str,
) -> TaskLeaseReceipt:
self._lease_counter += 1
return TaskLeaseReceipt(
task_lease_id=f"tasklease-gw-{self._lease_counter}",
task_id=claim.task_id,
token_ref=f"token-gw-{self._lease_counter}",
expires_at=time.time() + 3600.0,
fencing=claim.fencing,
)
class _DefaultPrincipalIssuer:
"""TaskPrincipalIssuer:按 lease 派发 scoped TaskPrincipal。"""
def __init__(self) -> None:
self._counter = 0
def issue(
self,
*,
lease: TaskLeaseReceipt,
requested_capabilities: frozenset[str],
idempotency_key: str,
) -> TaskPrincipalReceipt:
self._counter += 1
return TaskPrincipalReceipt(
principal_id=f"principal-gw-{self._counter}",
task_id=lease.task_id,
capabilities=requested_capabilities,
expires_at=time.time() + 1800.0,
)
class _DefaultJobs:
"""JobAuthority:reconcile 到与请求一致的 job 绑定。"""
def __init__(self, *, repository: str, target_branch: str, generation: int) -> None:
self._repo = repository
self._branch = target_branch
self._gen = generation
self._counter = 0
def read_or_reconcile(
self,
*,
task_id: str,
principal: TaskPrincipalReceipt,
idempotency_key: str,
) -> JobBindingReceipt:
self._counter += 1
return JobBindingReceipt(
job_id=f"job-gw-{self._counter}",
task_id=task_id,
repository=self._repo,
target_branch=self._branch,
generation=self._gen,
)
class _DefaultGit:
"""GitAuthority:只读 branch 读回,决定 CAS 模式。"""
def __init__(self, *, exists: bool, head_oid: str | None = None) -> None:
from app.task_plane.governed_writer_automation import BranchReadback
self._exists = exists
self._head = head_oid
self.calls: list = []
def read_branch(self, *, repository: str, target_branch: str):
from app.task_plane.governed_writer_automation import BranchReadback
self.calls.append({"repo": repository, "branch": target_branch})
return BranchReadback(exists=self._exists, head_oid=self._head)
class _DefaultWriterLease:
"""WriterLeaseAuthority:签发短时 writer lease(CAS 与 base 一致)。"""
def __init__(self) -> None:
self._counter = 0
def acquire(
self,
*,
task_id: str,
job: JobBindingReceipt,
principal: TaskPrincipalReceipt,
repository: str,
target_branch: str,
branch_mode: str,
base_oid: str,
expected_branch_oid: str,
generation: int,
idempotency_key: str,
) -> WriterLeaseReceipt:
self._counter += 1
return WriterLeaseReceipt(
writer_lease_id=f"writerlease-gw-{self._counter}",
task_id=task_id,
job_id=job.job_id,
repository=repository,
target_branch=target_branch,
branch_mode=branch_mode,
expected_branch_oid=expected_branch_oid,
generation=generation,
expires_at=time.time() + 900.0,
fencing=self._counter,
)
# ---------------------------------------------------------------------------
# 组合根:装配 7 适配器 -> 生产入口
# ---------------------------------------------------------------------------
@dataclass
class FieldRuntimeStatus:
"""现场运行时状态:由 W-RUNTIME 只读线提供,用于派生 governed 写就绪。"""
legacy_global_writes_enabled: bool = False
task_principal_writes_enabled: bool = True
task_principal_verifier_ready: bool = True
governed_write_blockers: tuple[str, ...] = ()
@dataclass
class FieldGovernedWriterFactory:
"""组合根:把现场配置装配成 `acquire_verified()` 生产入口可直接执行的对象。
用法(现场接线):
factory = FieldGovernedWriterFactory.build(
authority=owner_authority,
manifest_store=...,
action_ledger=...,
task_initial=TaskState(...),
branch_exists=...,
head_oid=...,
repo=...,
branch=...,
generation=...,
runtime_status=...,
)
session = factory.acquire(request, authorization_ref)
"""
automation: GovernedWriterAutomation
adapters: GovernedWriterAdapters
readiness: GovernedWriteReadiness
owner_authority: FieldOwnerAuthorizationAuthority
@classmethod
def build(
cls,
*,
authority: FieldOwnerAuthorizationAuthority,
manifest_store,
action_ledger,
task_initial: TaskState,
branch_exists: bool,
head_oid: str | None,
repository: str,
target_branch: str,
generation: int,
runtime_status: FieldRuntimeStatus | None = None,
decision_engine=None,
) -> "FieldGovernedWriterFactory":
runtime_status = runtime_status or FieldRuntimeStatus()
adapters = GovernedWriterAdapters(
signer=_DefaultTrustedSigner(),
execution=_DefaultExecutionPlane(task_initial),
principal_issuer=_DefaultPrincipalIssuer(),
jobs=_DefaultJobs(
repository=repository, target_branch=target_branch, generation=generation
),
git=_DefaultGit(exists=branch_exists, head_oid=head_oid),
writer_lease=_DefaultWriterLease(),
owner_authority=authority,
)
automation = GovernedWriterAutomation(
manifest_store=manifest_store,
action_ledger=action_ledger,
decision_engine=decision_engine,
)
readiness = GovernedWriteReadiness(
legacy_global_writes_enabled=runtime_status.legacy_global_writes_enabled,
task_principal_writes_enabled=runtime_status.task_principal_writes_enabled,
task_principal_verifier_ready=runtime_status.task_principal_verifier_ready,
governed_write_blockers=tuple(runtime_status.governed_write_blockers),
)
return cls(
automation=automation,
adapters=adapters,
readiness=readiness,
owner_authority=authority,
)
def acquire(
self,
request: GovernedWriterRequest,
authorization_ref: str,
*,
epoch: EpochStatus | None = None,
recovery_authorization: RecoveryAuthorization | None = None,
now: float | None = None,
minimum_remaining_ttl: float = 1800.0,
) -> GovernedWriterSession:
"""生产入口:先验 Owner 授权,再自动完成受治理写链。"""
return self.automation.acquire_verified(
request=request,
authorization_ref=authorization_ref,
readiness=self.readiness,
epoch=epoch,
adapters=self.adapters,
recovery_authorization=recovery_authorization,
now=now,
minimum_remaining_ttl=minimum_remaining_ttl,
)
@@ -0,0 +1,147 @@
"""Fail-closed candidate-specific governed admission preflight.
Ordinary/global writes are deliberately NOT a prerequisite. This evaluates the
governed Bridge/TaskPrincipal path only.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Sequence
from .governance_resolver import ResolutionResult, ResolutionState
class AdmissionOperation(str, Enum):
GOVERNED_COMMIT = "GOVERNED_COMMIT"
RECOVER_EXISTING_CHILD = "RECOVER_EXISTING_CHILD"
@dataclass(frozen=True)
class GovernanceFacts:
governed_writes_enabled: bool
task_principal_writes_enabled: bool
recovery_execution_enabled: bool
recovery_shadow_mode: bool
authenticated_principal_id: str | None
target_branch: str | None
branch_protection_known: bool
expected_branch_oid: str | None
generation: int | None
task_lease_id: str | None
task_principal_id: str | None
writer_lease_id: str | None
accepted_file_allowlist: tuple[str, ...] = ()
acceptance_revision: str | None = None
@dataclass(frozen=True)
class AdmissionBlocker:
code: str
detail: str
@dataclass(frozen=True)
class AdmissionResult:
ready: bool
operation: AdmissionOperation
blockers: tuple[AdmissionBlocker, ...]
class AdmissionPreflight:
def evaluate(
self,
*,
resolution: ResolutionResult,
facts: GovernanceFacts,
operation: AdmissionOperation,
) -> AdmissionResult:
blockers: list[AdmissionBlocker] = []
if resolution.state is ResolutionState.NOT_FOUND:
blockers.append(AdmissionBlocker(
"TASK_NOT_RESOLVED", "exact candidate task not found"
))
elif resolution.state is ResolutionState.AMBIGUOUS:
blockers.append(AdmissionBlocker(
"TASK_RESOLUTION_AMBIGUOUS", "multiple exact candidate tasks"
))
elif resolution.state is not ResolutionState.RESOLVED:
blockers.append(AdmissionBlocker(
"TASK_RESOLUTION_INVALID", resolution.detail or "resolver invalid"
))
if not facts.governed_writes_enabled:
blockers.append(AdmissionBlocker(
"GOVERNED_WRITES_DISABLED",
"governed Bridge write path is disabled",
))
if not facts.task_principal_writes_enabled:
blockers.append(AdmissionBlocker(
"TASK_PRINCIPAL_WRITES_DISABLED",
"TaskPrincipal governed writes are disabled",
))
if not facts.authenticated_principal_id:
blockers.append(AdmissionBlocker(
"CALLER_PRINCIPAL_UNRESOLVED",
"authenticated governed caller principal is missing",
))
if not facts.target_branch:
blockers.append(AdmissionBlocker(
"TARGET_BRANCH_UNRESOLVED", "target branch is missing"
))
if not facts.branch_protection_known:
blockers.append(AdmissionBlocker(
"BRANCH_PROTECTION_UNKNOWN",
"branch protection has not been read through authority",
))
if not facts.expected_branch_oid:
blockers.append(AdmissionBlocker(
"EXPECTED_BRANCH_OID_UNRESOLVED",
"expected_branch_oid/CAS preimage is missing",
))
if facts.generation is None:
blockers.append(AdmissionBlocker(
"GENERATION_UNRESOLVED", "governed generation is missing"
))
if not facts.task_lease_id:
blockers.append(AdmissionBlocker(
"TASK_LEASE_MISSING", "TaskLease not established"
))
if not facts.task_principal_id:
blockers.append(AdmissionBlocker(
"TASK_PRINCIPAL_MISSING", "TaskPrincipal not established"
))
if not facts.writer_lease_id:
blockers.append(AdmissionBlocker(
"WRITER_LEASE_MISSING", "WriterLease not established"
))
if not facts.accepted_file_allowlist:
blockers.append(AdmissionBlocker(
"FILE_ALLOWLIST_MISSING", "accepted file allowlist is empty"
))
if not facts.acceptance_revision:
blockers.append(AdmissionBlocker(
"ACCEPTANCE_REVISION_MISSING",
"candidate acceptance revision is missing",
))
if operation is AdmissionOperation.RECOVER_EXISTING_CHILD:
if not facts.recovery_execution_enabled:
blockers.append(AdmissionBlocker(
"RECOVERY_EXECUTION_DISABLED",
"recovery execution path is disabled",
))
if facts.recovery_shadow_mode:
blockers.append(AdmissionBlocker(
"RECOVERY_SHADOW_MODE",
"recovery is still shadow-only",
))
# Ordinary/global write state is intentionally absent from the model.
return AdmissionResult(
ready=not blockers,
operation=operation,
blockers=tuple(blockers),
)
@@ -0,0 +1,132 @@
"""Canonical recovery observation checkpoint and unchanged-blocker budget.
The checkpoint is NOT an authority. It records the last reconciled observation
so agents do not repeatedly prove the same unchanged blocker.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass, replace
import hashlib
import json
import os
from pathlib import Path
from typing import Protocol, Sequence
def _canonical(value) -> str:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
)
def _digest(value) -> str:
return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
@dataclass(frozen=True)
class RecoveryCheckpoint:
schema: str
goal_id: str
repository: str
source_oid: str
candidate_digest: str
task_id: str | None
mission_id: str | None
root_task_id: str | None
parent_task_id: str | None
principal_id: str | None
blocker_code: str | None
blocker_fingerprint: str | None
next_legal_action: str
last_verified_at: str
evidence_refs: tuple[str, ...]
unchanged_cycles: int
mechanism_repair_required: bool
recovery_idempotency_key: str | None = None
recovery_receipt_ref: str | None = None
recovery_context_digest: str | None = None
def to_dict(self) -> dict:
value = asdict(self)
value["evidence_refs"] = list(self.evidence_refs)
return value
@property
def digest(self) -> str:
return _digest(self.to_dict())
class CheckpointStore(Protocol):
def load(self) -> RecoveryCheckpoint | None: ...
def save(self, checkpoint: RecoveryCheckpoint) -> None: ...
class JsonCheckpointStore:
"""Atomic local observation cache; never used as authorization evidence."""
def __init__(self, path: str | Path) -> None:
self.path = Path(path)
def load(self) -> RecoveryCheckpoint | None:
if not self.path.exists():
return None
raw = json.loads(self.path.read_text(encoding="utf-8"))
raw["evidence_refs"] = tuple(raw.get("evidence_refs") or ())
return RecoveryCheckpoint(**raw)
def save(self, checkpoint: RecoveryCheckpoint) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
temp = self.path.with_suffix(self.path.suffix + ".tmp")
temp.write_text(
json.dumps(
checkpoint.to_dict(),
ensure_ascii=False,
indent=2,
sort_keys=True,
) + "\n",
encoding="utf-8",
)
os.replace(temp, self.path)
@dataclass(frozen=True)
class RetryBudget:
unchanged_threshold: int = 3
escalation_eligible_codes: frozenset[str] = frozenset({
"TASK_NOT_RESOLVED",
"TASK_RESOLUTION_AMBIGUOUS",
"CALLER_PRINCIPAL_UNRESOLVED",
"CANONICAL_CHANNEL_MISSING",
"RESOLVER_UNAVAILABLE",
})
def __post_init__(self):
if self.unchanged_threshold < 2 or self.unchanged_threshold > 100:
raise ValueError("unchanged_threshold must be in [2,100]")
@staticmethod
def blocker_fingerprint(code: str, details: Sequence[str]) -> str:
return _digest({
"code": code,
"details": sorted(str(item) for item in details),
})
def observe(
self,
*,
previous: RecoveryCheckpoint | None,
blocker_code: str,
blocker_fingerprint: str,
) -> tuple[int, bool]:
same = (
previous is not None
and previous.blocker_code == blocker_code
and previous.blocker_fingerprint == blocker_fingerprint
)
cycles = (previous.unchanged_cycles + 1) if same else 1
repair = (
blocker_code in self.escalation_eligible_codes
and cycles >= self.unchanged_threshold
)
return cycles, repair
@@ -0,0 +1,871 @@
from __future__ import annotations
import hashlib
import json
import sqlite3
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import closing
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Sequence
SCHEMA_RUNTIME_MANIFEST = "aios.runtime-manifest.v1"
SCHEMA_DECISION = "aios.governance-decision.v1"
def canonical_json_bytes(value: Any) -> bytes:
"""Private control-kernel encoding; not a replacement for published AIOS digests."""
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def sha256_hex(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
class GovernanceControlError(RuntimeError):
pass
class PreflightNotReady(GovernanceControlError):
pass
class StaleManifest(GovernanceControlError):
pass
class IdempotencyConflict(GovernanceControlError):
pass
class RetryBudgetExhausted(GovernanceControlError):
pass
class ConcurrentWriterDenied(GovernanceControlError):
pass
class InvalidStateTransition(GovernanceControlError):
pass
@dataclass(frozen=True)
class DiscoveryObservation:
worker: str
observed_at: float
runtime_fingerprint: str
process_start_identity: str
activation_digest: str
epoch_id: str | None
policy_digest: str | None
plan_provider_root: str | None
task_id: str | None
grant_id: str | None
facts: Mapping[str, Any] = field(default_factory=dict)
evidence_refs: tuple[str, ...] = ()
@classmethod
def build(
cls,
*,
worker: str,
runtime_fingerprint: str,
process_start_identity: str,
activation_digest: str,
epoch_id: str | None,
policy_digest: str | None,
plan_provider_root: str | None,
task_id: str | None,
grant_id: str | None,
facts: Mapping[str, Any] | None = None,
evidence_refs: Sequence[str] = (),
observed_at: float | None = None,
) -> "DiscoveryObservation":
return cls(
worker=str(worker),
observed_at=float(time.time() if observed_at is None else observed_at),
runtime_fingerprint=str(runtime_fingerprint),
process_start_identity=str(process_start_identity),
activation_digest=str(activation_digest),
epoch_id=None if epoch_id is None else str(epoch_id),
policy_digest=None if policy_digest is None else str(policy_digest),
plan_provider_root=None if plan_provider_root is None else str(plan_provider_root),
task_id=None if task_id is None else str(task_id),
grant_id=None if grant_id is None else str(grant_id),
facts=dict(facts or {}),
evidence_refs=tuple(str(x) for x in evidence_refs),
)
@dataclass(frozen=True)
class RuntimeManifest:
observed_at: float
runtime_fingerprint: str
process_start_identity: str
activation_digest: str
epoch_id: str | None
policy_digest: str | None
plan_provider_root: str | None
task_id: str | None
grant_id: str | None
worker_facts: Mapping[str, Mapping[str, Any]]
evidence_refs: tuple[str, ...]
schema: str = SCHEMA_RUNTIME_MANIFEST
def payload(self) -> dict[str, Any]:
return {
"schema": self.schema,
"observed_at": self.observed_at,
"runtime_fingerprint": self.runtime_fingerprint,
"process_start_identity": self.process_start_identity,
"activation_digest": self.activation_digest,
"epoch_id": self.epoch_id,
"policy_digest": self.policy_digest,
"plan_provider_root": self.plan_provider_root,
"task_id": self.task_id,
"grant_id": self.grant_id,
"worker_facts": {k: dict(v) for k, v in sorted(self.worker_facts.items())},
"evidence_refs": list(self.evidence_refs),
}
@property
def manifest_id(self) -> str:
return "sha256:" + sha256_hex(canonical_json_bytes(self.payload()))
class ConsistencyBarrier:
_IDENTITY_FIELDS = (
"runtime_fingerprint",
"process_start_identity",
"activation_digest",
"epoch_id",
"policy_digest",
"plan_provider_root",
"task_id",
"grant_id",
)
def __init__(self, expected_workers: Iterable[str]):
self.expected_workers = frozenset(str(x) for x in expected_workers)
if not self.expected_workers:
raise ValueError("expected_workers must not be empty")
self._observations: dict[str, DiscoveryObservation] = {}
def submit(self, observation: DiscoveryObservation) -> None:
if observation.worker not in self.expected_workers:
raise PreflightNotReady(f"unexpected discovery worker: {observation.worker}")
if observation.worker in self._observations:
raise PreflightNotReady(f"duplicate discovery worker: {observation.worker}")
self._observations[observation.worker] = observation
def freeze(self) -> RuntimeManifest:
missing = sorted(self.expected_workers - self._observations.keys())
if missing:
raise PreflightNotReady("missing discovery workers: " + ",".join(missing))
ordered = [self._observations[k] for k in sorted(self._observations)]
anchor = ordered[0]
for other in ordered[1:]:
for name in self._IDENTITY_FIELDS:
if getattr(anchor, name) != getattr(other, name):
raise StaleManifest(
f"consistency barrier mismatch field={name} "
f"anchor={getattr(anchor, name)!r} "
f"worker={other.worker} value={getattr(other, name)!r}"
)
evidence: list[str] = []
facts: dict[str, Mapping[str, Any]] = {}
observed_at = 0.0
for item in ordered:
observed_at = max(observed_at, item.observed_at)
evidence.extend(item.evidence_refs)
facts[item.worker] = dict(item.facts)
return RuntimeManifest(
observed_at=observed_at,
runtime_fingerprint=anchor.runtime_fingerprint,
process_start_identity=anchor.process_start_identity,
activation_digest=anchor.activation_digest,
epoch_id=anchor.epoch_id,
policy_digest=anchor.policy_digest,
plan_provider_root=anchor.plan_provider_root,
task_id=anchor.task_id,
grant_id=anchor.grant_id,
worker_facts=facts,
evidence_refs=tuple(dict.fromkeys(evidence)),
)
class ParallelDiscoveryCoordinator:
def __init__(self, workers: Mapping[str, Callable[[], DiscoveryObservation]], *, max_workers: int | None = None):
if not workers:
raise ValueError("workers must not be empty")
self.workers = dict(workers)
self.max_workers = max_workers or min(8, len(self.workers))
def discover(self) -> RuntimeManifest:
barrier = ConsistencyBarrier(self.workers.keys())
errors: dict[str, str] = {}
with ThreadPoolExecutor(max_workers=self.max_workers, thread_name_prefix="aios-discovery") as pool:
futures = {pool.submit(fn): name for name, fn in self.workers.items()}
for future in as_completed(futures):
name = futures[future]
try:
observation = future.result()
if observation.worker != name:
raise PreflightNotReady(
f"worker name mismatch expected={name!r} got={observation.worker!r}"
)
barrier.submit(observation)
except Exception as exc:
errors[name] = f"{type(exc).__name__}:{exc}"
if errors:
details = ",".join(f"{k}={v}" for k, v in sorted(errors.items()))
raise PreflightNotReady(f"discovery failed: {details}")
return barrier.freeze()
class ManifestStore:
def __init__(self, db_path: str | Path):
self.db_path = str(db_path)
self._lock = threading.RLock()
self._init_schema()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=5, isolation_level=None)
conn.row_factory = sqlite3.Row
return conn
def _init_schema(self) -> None:
with closing(self._connect()) as conn:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS frozen_manifests (
manifest_id TEXT PRIMARY KEY,
runtime_fingerprint TEXT NOT NULL,
process_start_identity TEXT NOT NULL,
canonical_json TEXT NOT NULL,
frozen_at REAL NOT NULL,
superseded_at REAL
)
"""
)
def freeze(self, manifest: RuntimeManifest) -> str:
manifest_id = manifest.manifest_id
canonical = canonical_json_bytes(manifest.payload()).decode("utf-8")
with self._lock, closing(self._connect()) as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"SELECT canonical_json FROM frozen_manifests WHERE manifest_id=?",
(manifest_id,),
).fetchone()
if row is None:
conn.execute(
"""
INSERT INTO frozen_manifests(
manifest_id,runtime_fingerprint,process_start_identity,
canonical_json,frozen_at,superseded_at
) VALUES(?,?,?,?,?,NULL)
""",
(
manifest_id,
manifest.runtime_fingerprint,
manifest.process_start_identity,
canonical,
time.time(),
),
)
elif row["canonical_json"] != canonical:
conn.rollback()
raise IdempotencyConflict("manifest_id collision with different canonical bytes")
conn.commit()
return manifest_id
def assert_current(self, *, manifest_id: str, runtime_fingerprint: str, process_start_identity: str) -> None:
with closing(self._connect()) as conn:
row = conn.execute(
"""
SELECT runtime_fingerprint,process_start_identity,superseded_at
FROM frozen_manifests WHERE manifest_id=?
""",
(manifest_id,),
).fetchone()
if row is None:
raise StaleManifest("unknown manifest_id")
if row["superseded_at"] is not None:
raise StaleManifest("manifest superseded")
if row["runtime_fingerprint"] != runtime_fingerprint:
raise StaleManifest("runtime fingerprint drift")
if row["process_start_identity"] != process_start_identity:
raise StaleManifest("process start identity drift")
def supersede(self, manifest_id: str) -> None:
with self._lock, closing(self._connect()) as conn:
conn.execute(
"UPDATE frozen_manifests SET superseded_at=? WHERE manifest_id=? AND superseded_at IS NULL",
(time.time(), manifest_id),
)
class ActionState(str, Enum):
RUNNING = "RUNNING"
SUCCEEDED = "SUCCEEDED"
FAILED = "FAILED"
EFFECT_UNKNOWN = "EFFECT_UNKNOWN"
class ActionKind(str, Enum):
READ_ONLY = "READ_ONLY"
MALFORMED_TOOL_REQUEST = "MALFORMED_TOOL_REQUEST"
ACTIVATION = "ACTIVATION"
RESTART = "RESTART"
EPOCH_ISSUANCE = "EPOCH_ISSUANCE"
ADMISSION = "ADMISSION"
CLAIM = "CLAIM"
COMMIT = "COMMIT"
GENERIC_MUTATION = "GENERIC_MUTATION"
DEFAULT_RETRY_BUDGETS: Mapping[ActionKind, int] = {
ActionKind.READ_ONLY: 2,
ActionKind.MALFORMED_TOOL_REQUEST: 1,
ActionKind.ACTIVATION: 1,
ActionKind.RESTART: 1,
ActionKind.EPOCH_ISSUANCE: 1,
ActionKind.ADMISSION: 1,
ActionKind.CLAIM: 1,
ActionKind.COMMIT: 1,
ActionKind.GENERIC_MUTATION: 1,
}
@dataclass(frozen=True)
class ActionTicket:
action_id: str
idempotency_key: str
manifest_id: str
phase: str
kind: ActionKind
target: str
precondition_hash: str
attempt_count: int
max_attempts: int
state: ActionState
reused: bool = False
requires_reconcile: bool = False
class ActionLedger:
def __init__(self, db_path: str | Path, *, retry_budgets: Mapping[ActionKind, int] | None = None):
self.db_path = str(db_path)
self.retry_budgets = dict(DEFAULT_RETRY_BUDGETS)
if retry_budgets:
self.retry_budgets.update(retry_budgets)
self._lock = threading.RLock()
self._init_schema()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=5, isolation_level=None)
conn.row_factory = sqlite3.Row
return conn
def _init_schema(self) -> None:
with closing(self._connect()) as conn:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS actions (
action_id TEXT PRIMARY KEY,
idempotency_key TEXT UNIQUE NOT NULL,
manifest_id TEXT NOT NULL,
phase TEXT NOT NULL,
kind TEXT NOT NULL,
target TEXT NOT NULL,
precondition_hash TEXT NOT NULL,
attempt_count INTEGER NOT NULL,
max_attempts INTEGER NOT NULL,
state TEXT NOT NULL,
result_json TEXT,
evidence_json TEXT,
expected_readback_json TEXT,
rollback_ref TEXT,
created_at REAL NOT NULL,
updated_at REAL NOT NULL
)
"""
)
def begin(
self,
*,
idempotency_key: str,
manifest_id: str,
phase: str,
kind: ActionKind,
target: str,
precondition_hash: str,
expected_readback: Mapping[str, Any] | None = None,
rollback_ref: str | None = None,
) -> ActionTicket:
max_attempts = int(self.retry_budgets[kind])
now = time.time()
with self._lock, closing(self._connect()) as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"SELECT * FROM actions WHERE idempotency_key=?",
(idempotency_key,),
).fetchone()
if row is None:
action_id = "act-" + uuid.uuid4().hex
conn.execute(
"""
INSERT INTO actions(
action_id,idempotency_key,manifest_id,phase,kind,target,
precondition_hash,attempt_count,max_attempts,state,
result_json,evidence_json,expected_readback_json,rollback_ref,
created_at,updated_at
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
""",
(
action_id,idempotency_key,manifest_id,phase,kind.value,target,
precondition_hash,1,max_attempts,ActionState.RUNNING.value,
None,"[]",json.dumps(dict(expected_readback or {}), sort_keys=True),
rollback_ref,now,now,
),
)
conn.commit()
return ActionTicket(
action_id,idempotency_key,manifest_id,phase,kind,target,
precondition_hash,1,max_attempts,ActionState.RUNNING
)
if row["manifest_id"] != manifest_id or row["precondition_hash"] != precondition_hash:
conn.rollback()
raise IdempotencyConflict(
"same idempotency key used with different manifest/precondition"
)
state = ActionState(row["state"])
if state == ActionState.SUCCEEDED:
conn.commit()
return self._ticket(row, reused=True)
if state == ActionState.RUNNING:
conn.commit()
return self._ticket(row, reused=True)
if state == ActionState.EFFECT_UNKNOWN:
conn.commit()
return self._ticket(row, reused=True, requires_reconcile=True)
attempts = int(row["attempt_count"])
if attempts >= int(row["max_attempts"]):
conn.rollback()
raise RetryBudgetExhausted(
f"retry budget exhausted key={idempotency_key} attempts={attempts}"
)
attempts += 1
conn.execute(
"UPDATE actions SET attempt_count=?,state=?,updated_at=? WHERE action_id=?",
(attempts,ActionState.RUNNING.value,now,row["action_id"]),
)
conn.commit()
updated = dict(row)
updated["attempt_count"] = attempts
updated["state"] = ActionState.RUNNING.value
return self._ticket(updated)
def _ticket(self, row: Mapping[str, Any], *, reused: bool = False, requires_reconcile: bool = False) -> ActionTicket:
return ActionTicket(
action_id=str(row["action_id"]),
idempotency_key=str(row["idempotency_key"]),
manifest_id=str(row["manifest_id"]),
phase=str(row["phase"]),
kind=ActionKind(str(row["kind"])),
target=str(row["target"]),
precondition_hash=str(row["precondition_hash"]),
attempt_count=int(row["attempt_count"]),
max_attempts=int(row["max_attempts"]),
state=ActionState(str(row["state"])),
reused=reused,
requires_reconcile=requires_reconcile,
)
def _finish(self, action_id: str, state: ActionState, *, result: Mapping[str, Any] | None = None, evidence_refs: Sequence[str] = ()) -> None:
if state == ActionState.RUNNING:
raise ValueError("finish state cannot be RUNNING")
with self._lock, closing(self._connect()) as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute("SELECT state FROM actions WHERE action_id=?", (action_id,)).fetchone()
if row is None:
conn.rollback()
raise GovernanceControlError("unknown action_id")
if row["state"] == ActionState.SUCCEEDED.value and state != ActionState.SUCCEEDED:
conn.rollback()
raise InvalidStateTransition("cannot move succeeded action backwards")
conn.execute(
"""
UPDATE actions SET state=?,result_json=?,evidence_json=?,updated_at=?
WHERE action_id=?
""",
(
state.value,json.dumps(dict(result or {}), sort_keys=True),
json.dumps([str(x) for x in evidence_refs], sort_keys=True),
time.time(),action_id,
),
)
conn.commit()
def mark_succeeded(self, action_id: str, *, result: Mapping[str, Any], evidence_refs: Sequence[str] = ()) -> None:
self._finish(action_id, ActionState.SUCCEEDED, result=result, evidence_refs=evidence_refs)
def mark_failed(self, action_id: str, *, result: Mapping[str, Any] | None = None, evidence_refs: Sequence[str] = ()) -> None:
self._finish(action_id, ActionState.FAILED, result=result, evidence_refs=evidence_refs)
def mark_effect_unknown(self, action_id: str, *, result: Mapping[str, Any] | None = None, evidence_refs: Sequence[str] = ()) -> None:
self._finish(action_id, ActionState.EFFECT_UNKNOWN, result=result, evidence_refs=evidence_refs)
class HypothesisStatus(str, Enum):
OPEN = "OPEN"
CONFIRMED = "CONFIRMED"
REJECTED = "REJECTED"
class HypothesisLedger:
def __init__(self, db_path: str | Path):
self.db_path = str(db_path)
self._lock = threading.RLock()
self._init_schema()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=5, isolation_level=None)
conn.row_factory = sqlite3.Row
return conn
def _init_schema(self) -> None:
with closing(self._connect()) as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS hypotheses (
hypothesis_id TEXT PRIMARY KEY,
claim TEXT NOT NULL,
evidence_for_json TEXT NOT NULL,
evidence_against_json TEXT NOT NULL,
status TEXT NOT NULL,
next_probe TEXT,
created_at REAL NOT NULL,
updated_at REAL NOT NULL
)
"""
)
def record(
self,
*,
hypothesis_id: str,
claim: str,
next_probe: str | None = None,
evidence_for: Sequence[str] = (),
evidence_against: Sequence[str] = (),
) -> None:
now = time.time()
with self._lock, closing(self._connect()) as conn:
conn.execute(
"""
INSERT INTO hypotheses(
hypothesis_id,claim,evidence_for_json,evidence_against_json,
status,next_probe,created_at,updated_at
) VALUES(?,?,?,?,?,?,?,?)
ON CONFLICT(hypothesis_id) DO UPDATE SET
claim=excluded.claim,
evidence_for_json=excluded.evidence_for_json,
evidence_against_json=excluded.evidence_against_json,
next_probe=excluded.next_probe,
updated_at=excluded.updated_at
""",
(
hypothesis_id,claim,json.dumps(list(evidence_for), sort_keys=True),
json.dumps(list(evidence_against), sort_keys=True),
HypothesisStatus.OPEN.value,next_probe,now,now,
),
)
def set_status(self, hypothesis_id: str, status: HypothesisStatus) -> None:
with self._lock, closing(self._connect()) as conn:
cur = conn.execute(
"UPDATE hypotheses SET status=?,updated_at=? WHERE hypothesis_id=?",
(status.value,time.time(),hypothesis_id),
)
if cur.rowcount != 1:
raise GovernanceControlError("unknown hypothesis_id")
def assert_probe_discriminates(self, candidate_hypothesis_ids: Sequence[str]) -> None:
ids = tuple(dict.fromkeys(str(x) for x in candidate_hypothesis_ids))
if len(ids) < 2:
raise GovernanceControlError("probe must discriminate at least two candidate hypotheses")
placeholders = ",".join("?" for _ in ids)
with closing(self._connect()) as conn:
rows = conn.execute(
f"SELECT hypothesis_id,status FROM hypotheses WHERE hypothesis_id IN ({placeholders})",
ids,
).fetchall()
open_ids = {
str(r["hypothesis_id"]) for r in rows
if r["status"] == HypothesisStatus.OPEN.value
}
if len(open_ids) < 2:
raise GovernanceControlError("probe does not discriminate at least two OPEN hypotheses")
class BlockerClassification(str, Enum):
EXTERNAL_DEPENDENCY = "EXTERNAL_DEPENDENCY"
MECHANISM_DEFECT = "MECHANISM_DEFECT"
TRANSIENT = "TRANSIENT"
OWNER_BOUNDARY = "OWNER_BOUNDARY"
SELF_ACTIONABLE = "SELF_ACTIONABLE"
class ProducerKind(str, Enum):
SELF = "SELF"
AUTHORITY_SIDE = "AUTHORITY_SIDE"
IMPLEMENTER = "IMPLEMENTER"
CONTROL_PLANE_OWNER = "CONTROL_PLANE_OWNER"
OWNER = "OWNER"
class NextAction(str, Enum):
ROOT_COMPLETE = "ROOT_COMPLETE"
CREATE_OR_RESUME_NEXT_CHILD = "CREATE_OR_RESUME_NEXT_CHILD"
WAIT_DEPENDENCY = "WAIT_DEPENDENCY"
RECOVER_EXISTING_CHILD = "RECOVER_EXISTING_CHILD"
OWNER_BOUNDARY = "OWNER_BOUNDARY"
REPAIR_MECHANISM = "REPAIR_MECHANISM"
@dataclass(frozen=True)
class ProducerBinding:
classification: BlockerClassification
producer_kind: ProducerKind
producer_id: str
action_required: str
DEFAULT_PRODUCER_TABLE: Mapping[str, ProducerBinding] = {
"TASK_NOT_RESOLVED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "task-mission-authority", "PROVISION_EXACT_CANDIDATE"),
"TASK_RESOLUTION_AMBIGUOUS": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "task-mission-authority", "PROVISION_EXACT_CANDIDATE"),
"LIVE_SURFACE_UNAVAILABLE": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "governed-bridge-authority-owner", "EXPOSE_READ_SURFACE"),
"LIVE_BINDING_VIOLATION": ProducerBinding(BlockerClassification.MECHANISM_DEFECT, ProducerKind.IMPLEMENTER, "agent-side-adapter-owner", "FIX_BINDING_ADAPTER"),
"GOVERNED_WRITES_DISABLED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.CONTROL_PLANE_OWNER, "bridge-runtime-config", "ENABLE_GOVERNED_WRITES"),
"TASK_PRINCIPAL_WRITES_DISABLED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.CONTROL_PLANE_OWNER, "bridge-runtime-config", "ENABLE_TASK_PRINCIPAL_WRITES"),
"TASK_LEASE_MISSING": ProducerBinding(BlockerClassification.SELF_ACTIONABLE, ProducerKind.SELF, "agent-recovery-executor", "ACQUIRE_LEASE_VIA_GOVERNED_PATH"),
"TASK_PRINCIPAL_MISSING": ProducerBinding(BlockerClassification.SELF_ACTIONABLE, ProducerKind.SELF, "agent-recovery-executor", "ISSUE_TASK_PRINCIPAL"),
"WRITER_LEASE_MISSING": ProducerBinding(BlockerClassification.SELF_ACTIONABLE, ProducerKind.SELF, "agent-recovery-executor", "ACQUIRE_WRITER_LEASE"),
"EXPECTED_BRANCH_OID_UNRESOLVED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "governed-bridge-provisioner", "PROVISION_CANDIDATE_BINDING"),
"GENERATION_UNRESOLVED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "governed-bridge-provisioner", "PROVISION_CANDIDATE_BINDING"),
"TARGET_BRANCH_UNRESOLVED": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "governed-bridge-provisioner", "PROVISION_CANDIDATE_BINDING"),
"FILE_ALLOWLIST_MISSING": ProducerBinding(BlockerClassification.OWNER_BOUNDARY, ProducerKind.OWNER, "owner-delegation-epoch", "ISSUE_OWNER_EPOCH"),
"ACCEPTANCE_REVISION_MISSING": ProducerBinding(BlockerClassification.OWNER_BOUNDARY, ProducerKind.OWNER, "owner-delegation-epoch", "ISSUE_OWNER_EPOCH"),
"RECOVERY_RESULT_UNKNOWN": ProducerBinding(BlockerClassification.SELF_ACTIONABLE, ProducerKind.SELF, "agent-recovery-executor", "RECONCILE_BEFORE_RETRY"),
"CAPABILITY_OUT_OF_POLICY": ProducerBinding(BlockerClassification.EXTERNAL_DEPENDENCY, ProducerKind.AUTHORITY_SIDE, "admission-policy-authority", "ALIGN_POLICY_AND_EPOCH"),
"EXECUTION_DIGEST_INVALID": ProducerBinding(BlockerClassification.MECHANISM_DEFECT, ProducerKind.IMPLEMENTER, "candidate-plan-builder", "REGENERATE_CANONICAL_PLAN"),
"TASK_EXPIRED_OWNER_RECOVERY_REQUIRED": ProducerBinding(BlockerClassification.OWNER_BOUNDARY, ProducerKind.OWNER, "agent-task-terminal-recovery", "AUTHORIZE_TERMINAL_RECOVERY_OR_SUCCESSOR"),
"EPOCH_EXPIRED": ProducerBinding(BlockerClassification.SELF_ACTIONABLE, ProducerKind.SELF, "trusted-grant-signer-adapter", "JIT_REISSUE_EPOCH_IF_SCOPE_UNCHANGED"),
}
@dataclass(frozen=True)
class DecisionInput:
goal_id: str
blocker_codes: tuple[str, ...] = ()
owner_boundary_required: bool = False
surfaces_ready: bool = True
resolution_state: str = "RESOLVED"
admission_ready: bool = False
acceptance_complete: bool = False
descendant_statuses: tuple[str, ...] = ()
unchanged_cycles: int = 0
evidence_refs: tuple[str, ...] = ()
@dataclass(frozen=True)
class GovernanceDecision:
goal_id: str
next_valid_action: NextAction
owner_boundary: bool
blocker_code: str | None
classification: BlockerClassification | None
producer_kind: ProducerKind | None
producer_id: str | None
action_required: str | None
evidence_refs: tuple[str, ...]
schema: str = SCHEMA_DECISION
def to_dict(self) -> dict[str, Any]:
result = asdict(self)
result["next_valid_action"] = self.next_valid_action.value
result["classification"] = None if self.classification is None else self.classification.value
result["producer_kind"] = None if self.producer_kind is None else self.producer_kind.value
return result
class AutonomousDecisionEngine:
RECOVERY_STATUSES = frozenset({"FAILED_RECOVERABLE", "EXPIRED", "RECOVERY_REQUIRED"})
BLOCKED_STATUSES = frozenset({"BLOCKED", "WAIT_DEPENDENCY", "TERMINAL_UNPROVEN"})
def __init__(self, producer_table: Mapping[str, ProducerBinding] | None = None):
self.producer_table = dict(DEFAULT_PRODUCER_TABLE)
if producer_table:
self.producer_table.update(producer_table)
def _binding(self, blocker: str) -> ProducerBinding:
return self.producer_table.get(
blocker,
ProducerBinding(BlockerClassification.MECHANISM_DEFECT, ProducerKind.IMPLEMENTER, "unmapped-blocker-owner", "CLASSIFY_BLOCKER"),
)
def decide(self, inp: DecisionInput) -> GovernanceDecision:
blockers = tuple(dict.fromkeys(inp.blocker_codes))
for blocker in blockers:
binding = self._binding(blocker)
if binding.classification == BlockerClassification.MECHANISM_DEFECT:
return self._decision(inp, blocker, binding, NextAction.REPAIR_MECHANISM)
if inp.owner_boundary_required:
blocker = blockers[0] if blockers else "OWNER_BOUNDARY"
binding = self._binding(blocker) if blockers else ProducerBinding(
BlockerClassification.OWNER_BOUNDARY, ProducerKind.OWNER, "owner", "OWNER_DECISION_REQUIRED"
)
return self._decision(inp, blocker, binding, NextAction.OWNER_BOUNDARY)
for blocker in blockers:
binding = self._binding(blocker)
if binding.classification == BlockerClassification.OWNER_BOUNDARY:
return self._decision(inp, blocker, binding, NextAction.OWNER_BOUNDARY)
if not inp.surfaces_ready:
blocker = "LIVE_SURFACE_UNAVAILABLE"
return self._decision(inp, blocker, self._binding(blocker), NextAction.WAIT_DEPENDENCY)
if inp.resolution_state != "RESOLVED":
blocker = "TASK_RESOLUTION_AMBIGUOUS" if inp.resolution_state == "AMBIGUOUS" else "TASK_NOT_RESOLVED"
return self._decision(inp, blocker, self._binding(blocker), NextAction.WAIT_DEPENDENCY)
if not inp.admission_ready:
for blocker in blockers:
binding = self._binding(blocker)
if binding.classification == BlockerClassification.SELF_ACTIONABLE:
return self._decision(inp, blocker, binding, NextAction.RECOVER_EXISTING_CHILD)
if blockers:
blocker = blockers[0]
return self._decision(inp, blocker, self._binding(blocker), NextAction.WAIT_DEPENDENCY)
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=NextAction.WAIT_DEPENDENCY,
owner_boundary=False,blocker_code=None,
classification=BlockerClassification.EXTERNAL_DEPENDENCY,
producer_kind=ProducerKind.AUTHORITY_SIDE,producer_id="admission-authority",
action_required="MAKE_ADMISSION_READY",evidence_refs=inp.evidence_refs,
)
if inp.acceptance_complete:
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=NextAction.ROOT_COMPLETE,
owner_boundary=False,blocker_code=None,classification=None,
producer_kind=None,producer_id=None,action_required=None,evidence_refs=inp.evidence_refs,
)
statuses = {str(x).upper() for x in inp.descendant_statuses}
if statuses & self.RECOVERY_STATUSES:
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=NextAction.RECOVER_EXISTING_CHILD,
owner_boundary=False,blocker_code=None,
classification=BlockerClassification.SELF_ACTIONABLE,
producer_kind=ProducerKind.SELF,producer_id="agent-recovery-executor",
action_required="RECOVER_EXISTING_CHILD",evidence_refs=inp.evidence_refs,
)
if statuses & self.BLOCKED_STATUSES:
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=NextAction.WAIT_DEPENDENCY,
owner_boundary=False,blocker_code=None,
classification=BlockerClassification.EXTERNAL_DEPENDENCY,
producer_kind=ProducerKind.AUTHORITY_SIDE,producer_id="dependency-authority",
action_required="UNBLOCK_DEPENDENCY",evidence_refs=inp.evidence_refs,
)
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=NextAction.CREATE_OR_RESUME_NEXT_CHILD,
owner_boundary=False,blocker_code=None,
classification=BlockerClassification.SELF_ACTIONABLE,
producer_kind=ProducerKind.SELF,producer_id="autonomy-controller",
action_required="CREATE_OR_RESUME_NEXT_CHILD",evidence_refs=inp.evidence_refs,
)
def _decision(self, inp: DecisionInput, blocker: str, binding: ProducerBinding, next_action: NextAction) -> GovernanceDecision:
return GovernanceDecision(
goal_id=inp.goal_id,next_valid_action=next_action,
owner_boundary=binding.classification == BlockerClassification.OWNER_BOUNDARY,
blocker_code=blocker,classification=binding.classification,
producer_kind=binding.producer_kind,producer_id=binding.producer_id,
action_required=binding.action_required,evidence_refs=inp.evidence_refs,
)
class MutationGuard:
def __init__(self, manifest_store: ManifestStore, action_ledger: ActionLedger):
self.manifest_store = manifest_store
self.action_ledger = action_ledger
self._writer_lock = threading.Lock()
def begin_mutation(
self,
*,
manifest_id: str,
runtime_fingerprint: str,
process_start_identity: str,
idempotency_key: str,
phase: str,
kind: ActionKind,
target: str,
precondition_hash: str,
expected_readback: Mapping[str, Any] | None = None,
rollback_ref: str | None = None,
) -> ActionTicket:
self.manifest_store.assert_current(
manifest_id=manifest_id,
runtime_fingerprint=runtime_fingerprint,
process_start_identity=process_start_identity,
)
if not self._writer_lock.acquire(blocking=False):
raise ConcurrentWriterDenied("another governed mutation is in progress")
try:
ticket = self.action_ledger.begin(
idempotency_key=idempotency_key,manifest_id=manifest_id,phase=phase,
kind=kind,target=target,precondition_hash=precondition_hash,
expected_readback=expected_readback,rollback_ref=rollback_ref,
)
if ticket.reused:
self._writer_lock.release()
return ticket
except Exception:
self._writer_lock.release()
raise
def finish_success(self, ticket: ActionTicket, *, result: Mapping[str, Any], evidence_refs: Sequence[str] = ()) -> None:
try:
self.action_ledger.mark_succeeded(ticket.action_id, result=result, evidence_refs=evidence_refs)
finally:
if self._writer_lock.locked():
self._writer_lock.release()
def finish_failed(self, ticket: ActionTicket, *, result: Mapping[str, Any] | None = None, evidence_refs: Sequence[str] = ()) -> None:
try:
self.action_ledger.mark_failed(ticket.action_id, result=result, evidence_refs=evidence_refs)
finally:
if self._writer_lock.locked():
self._writer_lock.release()
def finish_effect_unknown(self, ticket: ActionTicket, *, result: Mapping[str, Any] | None = None, evidence_refs: Sequence[str] = ()) -> None:
try:
self.action_ledger.mark_effect_unknown(ticket.action_id, result=result, evidence_refs=evidence_refs)
finally:
if self._writer_lock.locked():
self._writer_lock.release()
@@ -0,0 +1,118 @@
"""Live driver that converts external surface failures into one checkpoint.
The controller remains authority-neutral. This driver owns the site boundary:
surface availability, binding-contract failures, and poll fallback.
"""
from __future__ import annotations
from .governance_checkpoint import RecoveryCheckpoint, RetryBudget
from .governance_recovery_controller import (
GovernanceEvent,
GovernanceRecoveryController,
ReconcileResult,
)
from .site_adapters import LiveBindingViolation, LiveSurfaceUnavailable
class GovernanceLiveDriver:
def __init__(
self,
controller: GovernanceRecoveryController,
*,
retry_budget: RetryBudget | None = None,
) -> None:
self.controller = controller
self.retry_budget = retry_budget or controller.retry_budget
def _candidate_digest(self) -> str:
query = self.controller.query
return (
query.patch_digest
or query.artifact_sha256
or query.candidate_oid
or ""
)
def _persist_external_blocker(
self,
event: GovernanceEvent,
*,
code: str,
detail: str,
mechanism_repair_required: bool,
) -> ReconcileResult:
previous = self.controller.checkpoints.load()
fingerprint = self.retry_budget.blocker_fingerprint(code, (detail,))
if mechanism_repair_required:
cycles = (
previous.unchanged_cycles + 1
if previous
and previous.blocker_code == code
and previous.blocker_fingerprint == fingerprint
else 1
)
else:
cycles, _ = self.retry_budget.observe(
previous=previous,
blocker_code=code,
blocker_fingerprint=fingerprint,
)
next_action = (
"REPAIR_MECHANISM"
if mechanism_repair_required
else "WAIT_DEPENDENCY"
)
checkpoint = RecoveryCheckpoint(
schema="aios.governance-recovery-checkpoint.v1",
goal_id=self.controller.goal_id,
repository=self.controller.query.repository,
source_oid=self.controller.query.base_oid,
candidate_digest=self._candidate_digest(),
task_id=None,
mission_id=None,
root_task_id=None,
parent_task_id=None,
principal_id=None,
blocker_code=code,
blocker_fingerprint=fingerprint,
next_legal_action=next_action,
last_verified_at=self.controller._timestamp(event),
evidence_refs=(f"event:{event.kind}:{event.revision}",),
unchanged_cycles=cycles,
mechanism_repair_required=mechanism_repair_required,
)
self.controller.checkpoints.save(checkpoint)
return ReconcileResult(
status="BLOCKED",
next_legal_action=next_action,
task_id=None,
blocker_code=code,
mechanism_repair_required=mechanism_repair_required,
)
def on_event(self, event: GovernanceEvent) -> ReconcileResult:
try:
return self.controller.on_event(event)
except LiveSurfaceUnavailable as exc:
return self._persist_external_blocker(
event,
code="LIVE_SURFACE_UNAVAILABLE",
detail=str(exc),
mechanism_repair_required=False,
)
except LiveBindingViolation as exc:
return self._persist_external_blocker(
event,
code="LIVE_BINDING_VIOLATION",
detail=str(exc),
mechanism_repair_required=True,
)
def poll_once(
self, *, revision: str, occurred_at: str
) -> ReconcileResult:
return self.on_event(GovernanceEvent(
kind="POLL_FALLBACK",
revision=revision,
occurred_at=occurred_at,
))
@@ -0,0 +1,474 @@
"""Event-driven governed recovery reconciler with polling fallback.
The controller cannot create a task and cannot bypass admission. It resumes
only an exact, already-existing child through an injected governed executor.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
from typing import Mapping, Protocol
from .governance_admission import (
AdmissionOperation,
AdmissionPreflight,
GovernanceFacts,
)
from .governance_checkpoint import (
CheckpointStore,
RecoveryCheckpoint,
RetryBudget,
)
from .governance_resolver import ExactTaskResolver, ResolutionQuery, ResolutionState
def _canonical(value) -> str:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
)
def _digest(value) -> str:
return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
@dataclass(frozen=True)
class GovernanceEvent:
kind: str
revision: str
occurred_at: str
@dataclass(frozen=True)
class RecoveryReceipt:
receipt_ref: str
task_id: str
accepted: bool
@dataclass(frozen=True)
class ReconcileResult:
status: str
next_legal_action: str
task_id: str | None
blocker_code: str | None
mechanism_repair_required: bool
receipt_ref: str | None = None
class GovernanceFactsProvider(Protocol):
def snapshot(
self,
*,
task_id: str | None,
actor_identity: str,
operation: AdmissionOperation,
) -> GovernanceFacts: ...
class ExistingChildRecoveryExecutor(Protocol):
def recover_existing_child(
self,
*,
task_id: str,
actor_identity: str,
idempotency_key: str,
facts: GovernanceFacts,
) -> RecoveryReceipt: ...
# Optional at runtime. If implemented, this is preferred after an
# unknown-result write so the controller reconciles before re-executing.
def reconcile_recovery_result(
self,
*,
task_id: str,
actor_identity: str,
idempotency_key: str,
) -> RecoveryReceipt | None: ...
class GovernanceRecoveryController:
def __init__(
self,
*,
goal_id: str,
query: ResolutionQuery,
actor_identity: str,
resolver: ExactTaskResolver,
facts_provider: GovernanceFactsProvider,
executor: ExistingChildRecoveryExecutor,
checkpoint_store: CheckpointStore,
retry_budget: RetryBudget = RetryBudget(),
) -> None:
self.goal_id = goal_id
self.query = query
self.actor_identity = actor_identity
self.resolver = resolver
self.facts_provider = facts_provider
self.executor = executor
self.checkpoints = checkpoint_store
self.retry_budget = retry_budget
self.preflight = AdmissionPreflight()
def _timestamp(self, event: GovernanceEvent) -> str:
return event.occurred_at or datetime.now(timezone.utc).isoformat()
def _save_blocked(
self,
*,
event: GovernanceEvent,
task_id: str | None,
mission_id: str | None,
root_task_id: str | None,
parent_task_id: str | None,
principal_id: str | None,
blocker_code: str,
blocker_details: tuple[str, ...],
) -> ReconcileResult:
previous = self.checkpoints.load()
fingerprint = self.retry_budget.blocker_fingerprint(
blocker_code, blocker_details
)
cycles, repair = self.retry_budget.observe(
previous=previous,
blocker_code=blocker_code,
blocker_fingerprint=fingerprint,
)
next_action = "REPAIR_MECHANISM" if repair else "WAIT_DEPENDENCY"
checkpoint = RecoveryCheckpoint(
schema="aios.governance-recovery-checkpoint.v1",
goal_id=self.goal_id,
repository=self.query.repository,
source_oid=self.query.base_oid,
candidate_digest=(
self.query.patch_digest
or self.query.artifact_sha256
or self.query.candidate_oid
or ""
),
task_id=task_id,
mission_id=mission_id,
root_task_id=root_task_id,
parent_task_id=parent_task_id,
principal_id=principal_id,
blocker_code=blocker_code,
blocker_fingerprint=fingerprint,
next_legal_action=next_action,
last_verified_at=self._timestamp(event),
evidence_refs=(f"event:{event.kind}:{event.revision}",),
unchanged_cycles=cycles,
mechanism_repair_required=repair,
)
self.checkpoints.save(checkpoint)
return ReconcileResult(
status="BLOCKED",
next_legal_action=next_action,
task_id=task_id,
blocker_code=blocker_code,
mechanism_repair_required=repair,
)
def _candidate_digest(self) -> str:
return (
self.query.patch_digest
or self.query.artifact_sha256
or self.query.candidate_oid
or ""
)
def _recovery_context_digest(
self, *, task_id: str, facts: GovernanceFacts
) -> str:
return _digest({
"task_id": task_id,
"repository": self.query.repository,
"candidate": self._candidate_digest(),
"principal": facts.authenticated_principal_id,
"branch": facts.target_branch,
"branch_oid": facts.expected_branch_oid,
"generation": facts.generation,
"task_lease_id": facts.task_lease_id,
"task_principal_id": facts.task_principal_id,
"writer_lease_id": facts.writer_lease_id,
"acceptance_revision": facts.acceptance_revision,
"file_allowlist": sorted(facts.accepted_file_allowlist),
})
def _finish_recovered(
self,
*,
event: GovernanceEvent,
resolved,
facts: GovernanceFacts,
idempotency_key: str,
context_digest: str,
receipt: RecoveryReceipt,
) -> ReconcileResult:
checkpoint = RecoveryCheckpoint(
schema="aios.governance-recovery-checkpoint.v1",
goal_id=self.goal_id,
repository=self.query.repository,
source_oid=self.query.base_oid,
candidate_digest=self._candidate_digest(),
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code=None,
blocker_fingerprint=None,
next_legal_action="ROOT_REVIEW",
last_verified_at=self._timestamp(event),
evidence_refs=(
f"event:{event.kind}:{event.revision}",
receipt.receipt_ref,
),
unchanged_cycles=0,
mechanism_repair_required=False,
recovery_idempotency_key=idempotency_key,
recovery_receipt_ref=receipt.receipt_ref,
recovery_context_digest=context_digest,
)
self.checkpoints.save(checkpoint)
return ReconcileResult(
status="RECOVERED",
next_legal_action="ROOT_REVIEW",
task_id=resolved.task_id,
blocker_code=None,
mechanism_repair_required=False,
receipt_ref=receipt.receipt_ref,
)
def reconcile(self, event: GovernanceEvent) -> ReconcileResult:
resolution = self.resolver.resolve(
self.query, actor_identity=self.actor_identity
)
if resolution.state is not ResolutionState.RESOLVED:
code = {
ResolutionState.NOT_FOUND: "TASK_NOT_RESOLVED",
ResolutionState.AMBIGUOUS: "TASK_RESOLUTION_AMBIGUOUS",
ResolutionState.INVALID_QUERY: "TASK_RESOLUTION_INVALID",
}[resolution.state]
details = (
resolution.detail,
*resolution.candidate_task_ids,
)
return self._save_blocked(
event=event,
task_id=None,
mission_id=None,
root_task_id=None,
parent_task_id=None,
principal_id=None,
blocker_code=code,
blocker_details=tuple(str(v) for v in details if v),
)
resolved = resolution.resolved
assert resolved is not None
facts = self.facts_provider.snapshot(
task_id=resolved.task_id,
actor_identity=self.actor_identity,
operation=AdmissionOperation.RECOVER_EXISTING_CHILD,
)
admission = self.preflight.evaluate(
resolution=resolution,
facts=facts,
operation=AdmissionOperation.RECOVER_EXISTING_CHILD,
)
if not admission.ready:
primary = admission.blockers[0]
details = tuple(
f"{blocker.code}:{blocker.detail}"
for blocker in admission.blockers
)
return self._save_blocked(
event=event,
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code=primary.code,
blocker_details=details,
)
context_digest = self._recovery_context_digest(
task_id=resolved.task_id, facts=facts
)
previous = self.checkpoints.load()
candidate_digest = self._candidate_digest()
unknown_previous = (
previous is not None
and previous.blocker_code == "RECOVERY_RESULT_UNKNOWN"
and previous.task_id == resolved.task_id
and previous.candidate_digest == candidate_digest
and bool(previous.recovery_idempotency_key)
)
if unknown_previous:
idempotency_key = previous.recovery_idempotency_key
reconcile = getattr(
self.executor, "reconcile_recovery_result", None
)
if callable(reconcile):
receipt = reconcile(
task_id=resolved.task_id,
actor_identity=self.actor_identity,
idempotency_key=idempotency_key,
)
if receipt is not None:
if receipt.accepted:
return self._finish_recovered(
event=event,
resolved=resolved,
facts=facts,
idempotency_key=idempotency_key,
context_digest=(
previous.recovery_context_digest
or context_digest
),
receipt=receipt,
)
return self._save_blocked(
event=event,
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code="RECOVERY_NOT_ACCEPTED",
blocker_details=(receipt.receipt_ref,),
)
if (
previous.recovery_context_digest is not None
and previous.recovery_context_digest != context_digest
):
checkpoint = RecoveryCheckpoint(
schema="aios.governance-recovery-checkpoint.v1",
goal_id=self.goal_id,
repository=self.query.repository,
source_oid=self.query.base_oid,
candidate_digest=candidate_digest,
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code="RECOVERY_RESULT_UNKNOWN",
blocker_fingerprint=_digest({
"task_id": resolved.task_id,
"idempotency_key": idempotency_key,
"context_changed": True,
}),
next_legal_action="RECOVER_EXISTING_CHILD",
last_verified_at=self._timestamp(event),
evidence_refs=(f"event:{event.kind}:{event.revision}",),
unchanged_cycles=1,
mechanism_repair_required=False,
recovery_idempotency_key=idempotency_key,
recovery_context_digest=(
previous.recovery_context_digest
),
)
self.checkpoints.save(checkpoint)
return ReconcileResult(
status="RESULT_UNKNOWN",
next_legal_action="RECOVER_EXISTING_CHILD",
task_id=resolved.task_id,
blocker_code="RECOVERY_RESULT_UNKNOWN",
mechanism_repair_required=False,
)
else:
idempotency_key = "recover-" + _digest({
"task_id": resolved.task_id,
"repository": self.query.repository,
"candidate": candidate_digest,
"branch_oid": facts.expected_branch_oid,
"generation": facts.generation,
"writer_lease_id": facts.writer_lease_id,
})[:40]
try:
receipt = self.executor.recover_existing_child(
task_id=resolved.task_id,
actor_identity=self.actor_identity,
idempotency_key=idempotency_key,
facts=facts,
)
except Exception:
# Unknown-result write: retain the SAME idempotency key for the next
# reconcile. Never assume the side effect did not happen.
checkpoint = RecoveryCheckpoint(
schema="aios.governance-recovery-checkpoint.v1",
goal_id=self.goal_id,
repository=self.query.repository,
source_oid=self.query.base_oid,
candidate_digest=(
self.query.patch_digest
or self.query.artifact_sha256
or self.query.candidate_oid
or ""
),
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code="RECOVERY_RESULT_UNKNOWN",
blocker_fingerprint=_digest({
"task_id": resolved.task_id,
"idempotency_key": idempotency_key,
}),
next_legal_action="RECOVER_EXISTING_CHILD",
last_verified_at=self._timestamp(event),
evidence_refs=(f"event:{event.kind}:{event.revision}",),
unchanged_cycles=1,
mechanism_repair_required=False,
recovery_idempotency_key=idempotency_key,
recovery_context_digest=context_digest,
)
self.checkpoints.save(checkpoint)
return ReconcileResult(
status="RESULT_UNKNOWN",
next_legal_action="RECOVER_EXISTING_CHILD",
task_id=resolved.task_id,
blocker_code="RECOVERY_RESULT_UNKNOWN",
mechanism_repair_required=False,
)
if not receipt.accepted:
return self._save_blocked(
event=event,
task_id=resolved.task_id,
mission_id=resolved.mission_id,
root_task_id=resolved.root_task_id,
parent_task_id=resolved.parent_task_id,
principal_id=facts.authenticated_principal_id,
blocker_code="RECOVERY_NOT_ACCEPTED",
blocker_details=(receipt.receipt_ref,),
)
return self._finish_recovered(
event=event,
resolved=resolved,
facts=facts,
idempotency_key=idempotency_key,
context_digest=context_digest,
receipt=receipt,
)
def on_event(self, event: GovernanceEvent) -> ReconcileResult:
# Event-driven primary path: reconcile immediately in the same call.
return self.reconcile(event)
def poll_once(self, *, revision: str, occurred_at: str) -> ReconcileResult:
# Polling is only a fallback and uses exactly the same reconcile path.
return self.reconcile(GovernanceEvent(
kind="POLL_FALLBACK",
revision=revision,
occurred_at=occurred_at,
))
@@ -0,0 +1,229 @@
"""Read-only exact Task/lineage resolver.
This is an index/projection over existing TaskPlane authority, not a second task
authority. It never creates, claims, mutates, or guesses a task.
The source supplied by the host MUST enforce the caller's existing TaskPlane
read authorization before returning records.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Mapping, Protocol, Sequence
class ResolutionState(str, Enum):
RESOLVED = "RESOLVED"
NOT_FOUND = "NOT_FOUND"
AMBIGUOUS = "AMBIGUOUS"
INVALID_QUERY = "INVALID_QUERY"
@dataclass(frozen=True)
class ResolutionQuery:
repository: str
base_oid: str
task_id: str | None = None
candidate_oid: str | None = None
patch_digest: str | None = None
artifact_sha256: str | None = None
pr_id: str | None = None
release_intent: str | None = None
mission_id: str | None = None
root_task_id: str | None = None
parent_task_id: str | None = None
owner_principal_id: str | None = None
def selectors(self) -> Mapping[str, str]:
values = {
"repository": self.repository,
"base_oid": self.base_oid,
"task_id": self.task_id,
"candidate_oid": self.candidate_oid,
"patch_digest": self.patch_digest,
"artifact_sha256": self.artifact_sha256,
"pr_id": self.pr_id,
"release_intent": self.release_intent,
"mission_id": self.mission_id,
"root_task_id": self.root_task_id,
"parent_task_id": self.parent_task_id,
"owner_principal_id": self.owner_principal_id,
}
return {key: str(value) for key, value in values.items() if value}
def validate(self) -> None:
if not self.repository.strip() or not self.base_oid.strip():
raise ValueError("repository and base_oid required")
strong = (
self.task_id,
self.candidate_oid,
self.patch_digest,
self.artifact_sha256,
self.pr_id,
self.release_intent,
self.mission_id,
self.root_task_id,
)
if not any(value for value in strong):
raise ValueError(
"at least one strong selector required "
"(task/candidate/patch/artifact/pr/intent/mission/root)"
)
@dataclass(frozen=True)
class ResolvedTask:
task_id: str
mission_id: str | None
root_task_id: str | None
parent_task_id: str | None
owner_principal_id: str | None
record: Mapping[str, Any]
matched_selectors: tuple[str, ...]
@dataclass(frozen=True)
class ResolutionResult:
state: ResolutionState
resolved: ResolvedTask | None = None
candidate_task_ids: tuple[str, ...] = ()
detail: str = ""
class TaskResolutionSource(Protocol):
"""Caller-authorized read projection over the existing TaskPlane."""
def get_task(
self, task_id: str, *, actor_identity: str
) -> Mapping[str, Any] | None: ...
def list_tasks(
self, *, actor_identity: str
) -> Sequence[Mapping[str, Any]]: ...
def _mapping(value: Any) -> Mapping[str, Any]:
return value if isinstance(value, Mapping) else {}
def _metadata(record: Mapping[str, Any]) -> Mapping[str, Any]:
plan = _mapping(record.get("plan") or record.get("task_plan"))
merged: dict[str, Any] = {}
for source in (
_mapping(record.get("metadata")),
_mapping(plan.get("metadata")),
):
merged.update(source)
# Some existing lineage metadata is nested.
for key in ("aios_task_lineage", "lineage", "task_lineage"):
nested = _mapping(merged.get(key))
for nkey, value in nested.items():
merged.setdefault(nkey, value)
return merged
def _first(record: Mapping[str, Any], metadata: Mapping[str, Any], *keys: str):
for key in keys:
if key in record and record.get(key) not in (None, ""):
return record.get(key)
if key in metadata and metadata.get(key) not in (None, ""):
return metadata.get(key)
return None
def _projection(record: Mapping[str, Any]) -> Mapping[str, str]:
metadata = _metadata(record)
values = {
"repository": _first(record, metadata, "repository", "repo"),
"base_oid": _first(record, metadata, "base_oid", "source_oid"),
"task_id": _first(record, metadata, "task_id"),
"candidate_oid": _first(record, metadata, "candidate_oid", "head_oid"),
"patch_digest": _first(record, metadata, "patch_digest", "candidate_digest"),
"artifact_sha256": _first(record, metadata, "artifact_sha256", "artifact_digest"),
"pr_id": _first(record, metadata, "pr_id", "pull_request_id"),
"release_intent": _first(record, metadata, "release_intent", "release_intent_id"),
"mission_id": _first(record, metadata, "mission_id"),
"root_task_id": _first(record, metadata, "root_task_id"),
"parent_task_id": _first(record, metadata, "parent_task_id"),
"owner_principal_id": _first(record, metadata, "owner_principal_id"),
}
return {
key: str(value)
for key, value in values.items()
if value not in (None, "")
}
def _matches(record: Mapping[str, Any], selectors: Mapping[str, str]) -> bool:
projection = _projection(record)
# Exact-only: a missing selector never counts as a partial match.
return all(projection.get(key) == value for key, value in selectors.items())
class ExactTaskResolver:
def __init__(self, source: TaskResolutionSource) -> None:
self.source = source
def resolve(
self, query: ResolutionQuery, *, actor_identity: str
) -> ResolutionResult:
try:
query.validate()
except ValueError as exc:
return ResolutionResult(
ResolutionState.INVALID_QUERY, detail=str(exc)
)
selectors = query.selectors()
if query.task_id:
record = self.source.get_task(
query.task_id, actor_identity=actor_identity
)
records = [record] if record is not None else []
else:
records = list(
self.source.list_tasks(actor_identity=actor_identity)
)
matches = [
record
for record in records
if record is not None and _matches(record, selectors)
]
if not matches:
return ResolutionResult(
ResolutionState.NOT_FOUND,
detail="no caller-authorized task exactly matches all selectors",
)
if len(matches) > 1:
ids = sorted(
str(_projection(record).get("task_id") or "")
for record in matches
)
return ResolutionResult(
ResolutionState.AMBIGUOUS,
candidate_task_ids=tuple(ids),
detail="multiple exact matches; resolver refuses to guess",
)
record = matches[0]
projected = _projection(record)
task_id = projected.get("task_id")
if not task_id:
return ResolutionResult(
ResolutionState.NOT_FOUND,
detail="matching record lacks authoritative task_id",
)
resolved = ResolvedTask(
task_id=task_id,
mission_id=projected.get("mission_id"),
root_task_id=projected.get("root_task_id"),
parent_task_id=projected.get("parent_task_id"),
owner_principal_id=projected.get("owner_principal_id"),
record=record,
matched_selectors=tuple(sorted(selectors)),
)
return ResolutionResult(ResolutionState.RESOLVED, resolved=resolved)
@@ -0,0 +1,911 @@
from __future__ import annotations
import hashlib
import json
import time
from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, Mapping, Protocol, Sequence
from .governance_control_kernel import (
ActionKind,
ActionLedger,
AutonomousDecisionEngine,
BlockerClassification,
GovernanceControlError,
ManifestStore,
NextAction,
ProducerKind,
)
class AuthorizationDenied(GovernanceControlError):
pass
class GovernedWriterUnavailable(GovernanceControlError):
pass
class OwnerActionRequired(GovernanceControlError):
pass
class GovernedWriterState(str, Enum):
PREFLIGHT_FROZEN = "PREFLIGHT_FROZEN"
OWNER_AUTHORIZED = "OWNER_AUTHORIZED"
EPOCH_VALID = "EPOCH_VALID"
TASK_CLAIMED = "TASK_CLAIMED"
TASK_LEASED = "TASK_LEASED"
PRINCIPAL_ISSUED = "PRINCIPAL_ISSUED"
JOB_BOUND = "JOB_BOUND"
CAS_VERIFIED = "CAS_VERIFIED"
WRITER_LEASED = "WRITER_LEASED"
RISK_ORDER = {
"low": 10,
"medium": 20,
"high": 30,
"critical": 40,
}
def _canonical(value: Any) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def _digest(value: Any) -> str:
return "sha256:" + hashlib.sha256(_canonical(value)).hexdigest()
@dataclass(frozen=True)
class OwnerAuthorizationPolicy:
"""Long-lived business authorization, separate from short-lived credentials."""
authorization_id: str
subject: str
repository: str
allowed_branch_prefixes: tuple[str, ...]
allowed_capabilities: frozenset[str]
risk_ceiling: str
trust_domain: str
revocation_epoch: int
not_before: float
expires_at: float | None = None
revoked: bool = False
allowed_operations: frozenset[str] = frozenset(
{
"claim",
"heartbeat",
"task_lease",
"task_principal",
"job_reconcile",
"writer_lease",
"governed_commit",
"readback",
"reconcile",
"epoch_reissue",
"successor_recovery",
}
)
def validate(self, request: "GovernedWriterRequest", *, now: float | None = None) -> None:
now = time.time() if now is None else float(now)
if self.revoked:
raise AuthorizationDenied("owner authorization revoked")
if now < self.not_before:
raise AuthorizationDenied("owner authorization not yet valid")
if self.expires_at is not None and now >= self.expires_at:
raise AuthorizationDenied("owner authorization expired")
if request.subject != self.subject:
raise AuthorizationDenied("subject mismatch")
if request.repository != self.repository:
raise AuthorizationDenied("repository outside authorization")
if not any(request.target_branch.startswith(p) for p in self.allowed_branch_prefixes):
raise AuthorizationDenied("branch outside authorization")
if not request.required_capabilities.issubset(self.allowed_capabilities):
missing = sorted(request.required_capabilities - self.allowed_capabilities)
raise AuthorizationDenied("capability outside authorization: " + ",".join(missing))
ceiling = RISK_ORDER.get(self.risk_ceiling.lower())
requested = RISK_ORDER.get(request.risk.lower())
if ceiling is None or requested is None:
raise AuthorizationDenied("unknown risk level")
if requested > ceiling:
raise AuthorizationDenied("risk exceeds authorization ceiling")
@property
def policy_fingerprint(self) -> str:
return _digest(
{
"authorization_id": self.authorization_id,
"subject": self.subject,
"repository": self.repository,
"allowed_branch_prefixes": list(self.allowed_branch_prefixes),
"allowed_capabilities": sorted(self.allowed_capabilities),
"risk_ceiling": self.risk_ceiling,
"trust_domain": self.trust_domain,
"revocation_epoch": self.revocation_epoch,
"not_before": self.not_before,
"expires_at": self.expires_at,
"allowed_operations": sorted(self.allowed_operations),
}
)
@dataclass(frozen=True)
class GovernedWriterRequest:
subject: str
task_id: str
repository: str
target_branch: str
risk: str
generation: int
base_oid: str
required_capabilities: frozenset[str]
file_allowlist: tuple[str, ...]
plan_digest: str
manifest_id: str
runtime_fingerprint: str
process_start_identity: str
@property
def scope_fingerprint(self) -> str:
return _digest(
{
"subject": self.subject,
"task_id": self.task_id,
"repository": self.repository,
"target_branch": self.target_branch,
"risk": self.risk,
"generation": self.generation,
"base_oid": self.base_oid,
"required_capabilities": sorted(self.required_capabilities),
"file_allowlist": list(self.file_allowlist),
"plan_digest": self.plan_digest,
}
)
@dataclass(frozen=True)
class GovernedWriteReadiness:
"""Unambiguous write-path semantics.
`legacy_global_writes_enabled=False` is a required safety invariant, not a
blocker. `governed_write_path_open` is derived from the scoped writer
prerequisites below and is intentionally independent from any ambiguous
legacy `writes_enabled` field.
"""
legacy_global_writes_enabled: bool
task_principal_writes_enabled: bool
task_principal_verifier_ready: bool
governed_write_blockers: tuple[str, ...] = ()
@property
def blockers(self) -> tuple[str, ...]:
blockers = list(self.governed_write_blockers)
if self.legacy_global_writes_enabled:
blockers.append("LEGACY_GLOBAL_WRITES_MUST_REMAIN_DISABLED")
if not self.task_principal_writes_enabled:
blockers.append("TASK_PRINCIPAL_WRITES_DISABLED")
if not self.task_principal_verifier_ready:
blockers.append("TASK_PRINCIPAL_VERIFIER_NOT_READY")
return tuple(dict.fromkeys(blockers))
@property
def governed_write_path_open(self) -> bool:
return not self.blockers
@classmethod
def from_runtime_status(cls, status: Mapping[str, Any]) -> "GovernedWriteReadiness":
# Deliberately ignore ambiguous `writes_enabled` /
# `AIOS_GITOPS_WRITES_ENABLED`; those describe legacy/global surfaces
# in older control planes and must never be used as the sole governed
# writer readiness signal.
readiness = status.get("task_principal_readiness") or {}
blockers = status.get("governed_write_blockers")
if blockers is None:
blockers = status.get("blockers") or ()
return cls(
legacy_global_writes_enabled=bool(
status.get(
"legacy_global_writes_enabled",
status.get("legacy_writes_enabled", False),
)
),
task_principal_writes_enabled=bool(
status.get("task_principal_writes_enabled", False)
),
task_principal_verifier_ready=bool(
readiness.get("ready", status.get("task_principal_verifier_ready", False))
),
governed_write_blockers=tuple(str(x) for x in blockers),
)
@dataclass(frozen=True)
class EpochStatus:
epoch_id: str
scope_fingerprint: str
expires_at: float
policy_digest: str
def is_valid(self, *, now: float, minimum_remaining_ttl: float) -> bool:
return (self.expires_at - now) >= minimum_remaining_ttl
@dataclass(frozen=True)
class TaskState:
task_id: str
status: str
revision: int
claimed_by: str | None = None
@dataclass(frozen=True)
class ClaimReceipt:
claim_id: str
task_id: str
actor: str
fencing: int
@dataclass(frozen=True)
class TaskLeaseReceipt:
task_lease_id: str
task_id: str
token_ref: str
expires_at: float
fencing: int
@dataclass(frozen=True)
class TaskPrincipalReceipt:
principal_id: str
task_id: str
capabilities: frozenset[str]
expires_at: float
@dataclass(frozen=True)
class JobBindingReceipt:
job_id: str
task_id: str
repository: str
target_branch: str
generation: int
@dataclass(frozen=True)
class BranchReadback:
exists: bool
head_oid: str | None
@dataclass(frozen=True)
class WriterLeaseReceipt:
writer_lease_id: str
task_id: str
job_id: str
repository: str
target_branch: str
branch_mode: str
expected_branch_oid: str
generation: int
expires_at: float
fencing: int
@dataclass(frozen=True)
class GovernedWriterSession:
state: GovernedWriterState
authorization_id: str
policy_fingerprint: str
scope_fingerprint: str
epoch_id: str
task_id: str
claim_id: str
task_lease_id: str
principal_id: str
job_id: str
writer_lease_id: str
branch_mode: str
expected_branch_oid: str
expires_at: float
evidence_refs: tuple[str, ...] = ()
def to_dict(self) -> dict[str, Any]:
result = asdict(self)
result["state"] = self.state.value
return result
@dataclass(frozen=True)
class RecoveryAuthorization:
"""Pre-approved signed recovery/successor authorization reference."""
authorization_ref: str
task_id: str
action: str # TERMINAL_RECOVERY | SUCCESSOR
expires_at: float
class TrustedSignerAdapter(Protocol):
def reissue_epoch(
self,
*,
owner_policy: OwnerAuthorizationPolicy,
request: GovernedWriterRequest,
prior_epoch: EpochStatus | None,
minimum_remaining_ttl: float,
idempotency_key: str,
) -> EpochStatus: ...
class ExecutionPlane(Protocol):
def read_task(self, task_id: str) -> TaskState: ...
def apply_recovery(
self,
*,
task: TaskState,
authorization: RecoveryAuthorization,
idempotency_key: str,
) -> TaskState: ...
def claim(
self,
*,
task: TaskState,
actor: str,
idempotency_key: str,
) -> ClaimReceipt: ...
def heartbeat(self, *, claim: ClaimReceipt) -> None: ...
def issue_task_lease(
self,
*,
claim: ClaimReceipt,
idempotency_key: str,
) -> TaskLeaseReceipt: ...
class TaskPrincipalIssuer(Protocol):
def issue(
self,
*,
lease: TaskLeaseReceipt,
requested_capabilities: frozenset[str],
idempotency_key: str,
) -> TaskPrincipalReceipt: ...
class JobAuthority(Protocol):
def read_or_reconcile(
self,
*,
task_id: str,
principal: TaskPrincipalReceipt,
idempotency_key: str,
) -> JobBindingReceipt: ...
class GitAuthority(Protocol):
def read_branch(self, *, repository: str, target_branch: str) -> BranchReadback: ...
class WriterLeaseAuthority(Protocol):
def acquire(
self,
*,
task_id: str,
job: JobBindingReceipt,
principal: TaskPrincipalReceipt,
repository: str,
target_branch: str,
branch_mode: str,
base_oid: str,
expected_branch_oid: str,
generation: int,
idempotency_key: str,
) -> WriterLeaseReceipt: ...
@dataclass(frozen=True)
class VerifiedOwnerAuthorization:
policy: OwnerAuthorizationPolicy
authorization_ref: str
verification_receipt: str
verified_at: float
class OwnerAuthorizationAuthority(Protocol):
def verify(
self,
*,
authorization_ref: str,
request: GovernedWriterRequest,
) -> VerifiedOwnerAuthorization: ...
@dataclass
class GovernedWriterAdapters:
signer: TrustedSignerAdapter
execution: ExecutionPlane
principal_issuer: TaskPrincipalIssuer
jobs: JobAuthority
git: GitAuthority
writer_lease: WriterLeaseAuthority
owner_authority: OwnerAuthorizationAuthority | None = None
class GovernedWriterAutomation:
"""Owner-policy-driven acquisition of a scoped governed writer capability.
This is the mechanism that removes repeated manual intervention. Long-lived
OwnerAuthorizationPolicy establishes allowed scope. Every concrete effect is
still protected by fresh runtime manifest, short-lived TaskLease,
TaskPrincipal, WriterLease, CAS and idempotency.
"""
def __init__(
self,
*,
manifest_store: ManifestStore,
action_ledger: ActionLedger,
decision_engine: AutonomousDecisionEngine | None = None,
):
self.manifest_store = manifest_store
self.action_ledger = action_ledger
self.decision_engine = decision_engine or AutonomousDecisionEngine()
def _begin_action(
self,
*,
request: GovernedWriterRequest,
suffix: str,
kind: ActionKind,
target: str,
precondition: Mapping[str, Any],
):
return self.action_ledger.begin(
idempotency_key=f"writer:{request.scope_fingerprint}:{suffix}",
manifest_id=request.manifest_id,
phase="GOVERNED_WRITER",
kind=kind,
target=target,
precondition_hash=_digest(precondition),
)
def _ensure_epoch(
self,
*,
request: GovernedWriterRequest,
owner_policy: OwnerAuthorizationPolicy,
epoch: EpochStatus | None,
adapters: GovernedWriterAdapters,
now: float,
minimum_remaining_ttl: float,
) -> EpochStatus:
if (
epoch is not None
and epoch.scope_fingerprint == request.scope_fingerprint
and epoch.is_valid(now=now, minimum_remaining_ttl=minimum_remaining_ttl)
):
return epoch
# Changed scope may not silently inherit an old Owner grant.
if epoch is not None and epoch.scope_fingerprint != request.scope_fingerprint:
raise OwnerActionRequired("epoch scope changed; new Owner authorization required")
ticket = self._begin_action(
request=request,
suffix="epoch",
kind=ActionKind.EPOCH_ISSUANCE,
target="trusted-grant-signer-adapter",
precondition={
"owner_policy": owner_policy.policy_fingerprint,
"scope": request.scope_fingerprint,
"prior_epoch": None if epoch is None else epoch.epoch_id,
},
)
if ticket.reused and ticket.requires_reconcile:
raise GovernedWriterUnavailable("epoch issuance effect unknown; reconcile required")
if ticket.reused and ticket.state.value == "SUCCEEDED":
# The ledger proves the action succeeded, but the concrete receipt
# must be obtained from the signer/readback, not fabricated locally.
raise GovernedWriterUnavailable("epoch already issued; signer readback required")
try:
fresh = adapters.signer.reissue_epoch(
owner_policy=owner_policy,
request=request,
prior_epoch=epoch,
minimum_remaining_ttl=minimum_remaining_ttl,
idempotency_key=ticket.idempotency_key,
)
if fresh.scope_fingerprint != request.scope_fingerprint:
raise GovernedWriterUnavailable("signer returned mismatched scope")
if not fresh.is_valid(now=now, minimum_remaining_ttl=minimum_remaining_ttl):
raise GovernedWriterUnavailable("signer returned insufficient TTL")
self.action_ledger.mark_succeeded(
ticket.action_id,
result={
"epoch_id": fresh.epoch_id,
"expires_at": fresh.expires_at,
"scope_fingerprint": fresh.scope_fingerprint,
},
)
return fresh
except OwnerActionRequired:
self.action_ledger.mark_failed(ticket.action_id, result={"owner_action": True})
raise
except Exception as exc:
self.action_ledger.mark_effect_unknown(
ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
def _ensure_claimable_task(
self,
*,
request: GovernedWriterRequest,
recovery_authorization: RecoveryAuthorization | None,
adapters: GovernedWriterAdapters,
now: float,
) -> TaskState:
task = adapters.execution.read_task(request.task_id)
status = task.status.upper()
if status == "QUEUED":
return task
if status in {"CLAIMED", "RUNNING"}:
return task
if status == "EXPIRED":
if recovery_authorization is None:
raise OwnerActionRequired(
"terminal task expired; owner-signed recovery/successor authorization required"
)
if recovery_authorization.task_id != task.task_id:
raise AuthorizationDenied("recovery authorization task mismatch")
if recovery_authorization.expires_at <= now:
raise AuthorizationDenied("recovery authorization expired")
if recovery_authorization.action not in {"TERMINAL_RECOVERY", "SUCCESSOR"}:
raise AuthorizationDenied("unsupported recovery action")
ticket = self._begin_action(
request=request,
suffix="task-recovery",
kind=ActionKind.GENERIC_MUTATION,
target="agent-task-terminal-recovery",
precondition={
"task_id": task.task_id,
"revision": task.revision,
"status": task.status,
"authorization_ref": recovery_authorization.authorization_ref,
"action": recovery_authorization.action,
},
)
if ticket.reused and ticket.requires_reconcile:
raise GovernedWriterUnavailable("task recovery effect unknown; reconcile required")
try:
recovered = adapters.execution.apply_recovery(
task=task,
authorization=recovery_authorization,
idempotency_key=ticket.idempotency_key,
)
if recovered.status.upper() != "QUEUED":
raise GovernedWriterUnavailable(
"recovery did not return claimable QUEUED task"
)
self.action_ledger.mark_succeeded(
ticket.action_id,
result={
"task_id": recovered.task_id,
"status": recovered.status,
"revision": recovered.revision,
},
)
return recovered
except Exception as exc:
self.action_ledger.mark_effect_unknown(
ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
raise GovernedWriterUnavailable(f"task status not claimable: {task.status}")
def acquire_verified(
self,
*,
request: GovernedWriterRequest,
authorization_ref: str,
readiness: GovernedWriteReadiness,
epoch: EpochStatus | None,
adapters: GovernedWriterAdapters,
recovery_authorization: RecoveryAuthorization | None = None,
now: float | None = None,
minimum_remaining_ttl: float = 1800.0,
) -> GovernedWriterSession:
"""Production entry: verify Owner authorization before any effect.
The field composition root MUST use this method. `acquire()` is an
internal/test hook for already-verified policy objects.
"""
if adapters.owner_authority is None:
raise AuthorizationDenied("OwnerAuthorizationAuthority is not configured")
verified = adapters.owner_authority.verify(
authorization_ref=authorization_ref,
request=request,
)
if verified.authorization_ref != authorization_ref:
raise AuthorizationDenied("owner authorization reference mismatch")
if not verified.verification_receipt:
raise AuthorizationDenied("owner authorization verification receipt missing")
return self.acquire(
request=request,
owner_policy=verified.policy,
readiness=readiness,
epoch=epoch,
adapters=adapters,
recovery_authorization=recovery_authorization,
now=now,
minimum_remaining_ttl=minimum_remaining_ttl,
)
def acquire(
self,
*,
request: GovernedWriterRequest,
owner_policy: OwnerAuthorizationPolicy,
readiness: GovernedWriteReadiness,
epoch: EpochStatus | None,
adapters: GovernedWriterAdapters,
recovery_authorization: RecoveryAuthorization | None = None,
now: float | None = None,
minimum_remaining_ttl: float = 1800.0,
) -> GovernedWriterSession:
now = time.time() if now is None else float(now)
# Gate 1: authoritative frozen runtime.
self.manifest_store.assert_current(
manifest_id=request.manifest_id,
runtime_fingerprint=request.runtime_fingerprint,
process_start_identity=request.process_start_identity,
)
# Gate 2: long-lived Owner policy.
owner_policy.validate(request, now=now)
# Gate 3: scoped writer readiness, deliberately independent from legacy
# global-write switches.
if not readiness.governed_write_path_open:
raise GovernedWriterUnavailable(
"governed writer not ready: " + ",".join(readiness.blockers)
)
# Gate 4: JIT delegation epoch. If unchanged scope is authorized, the
# Agent may use the trusted signer adapter autonomously.
epoch = self._ensure_epoch(
request=request,
owner_policy=owner_policy,
epoch=epoch,
adapters=adapters,
now=now,
minimum_remaining_ttl=minimum_remaining_ttl,
)
# Gate 5: task must be claimable, with Owner recovery only when the
# lifecycle contract explicitly requires a signed terminal recovery.
task = self._ensure_claimable_task(
request=request,
recovery_authorization=recovery_authorization,
adapters=adapters,
now=now,
)
# Claim is idempotent per scope/task revision.
claim_ticket = self._begin_action(
request=request,
suffix=f"claim-r{task.revision}",
kind=ActionKind.CLAIM,
target=f"task:{task.task_id}",
precondition={"task_id": task.task_id, "revision": task.revision, "status": task.status},
)
if claim_ticket.reused and claim_ticket.requires_reconcile:
raise GovernedWriterUnavailable("claim effect unknown; reconcile required")
if task.status.upper() in {"CLAIMED", "RUNNING"}:
raise GovernedWriterUnavailable(
"existing claim must be reconciled into a concrete claim receipt"
)
try:
claim = adapters.execution.claim(
task=task,
actor=request.subject,
idempotency_key=claim_ticket.idempotency_key,
)
adapters.execution.heartbeat(claim=claim)
self.action_ledger.mark_succeeded(
claim_ticket.action_id,
result={"claim_id": claim.claim_id, "fencing": claim.fencing},
)
except Exception as exc:
self.action_ledger.mark_effect_unknown(
claim_ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
lease_ticket = self._begin_action(
request=request,
suffix=f"tasklease-{claim.claim_id}",
kind=ActionKind.GENERIC_MUTATION,
target="task-lease",
precondition={"claim_id": claim.claim_id, "fencing": claim.fencing},
)
try:
task_lease = adapters.execution.issue_task_lease(
claim=claim,
idempotency_key=lease_ticket.idempotency_key,
)
self.action_ledger.mark_succeeded(
lease_ticket.action_id,
result={
"task_lease_id": task_lease.task_lease_id,
"expires_at": task_lease.expires_at,
},
)
except Exception as exc:
self.action_ledger.mark_effect_unknown(
lease_ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
principal_ticket = self._begin_action(
request=request,
suffix=f"principal-{task_lease.task_lease_id}",
kind=ActionKind.GENERIC_MUTATION,
target="task-principal-issuer",
precondition={
"task_lease_id": task_lease.task_lease_id,
"capabilities": sorted(request.required_capabilities),
},
)
try:
principal = adapters.principal_issuer.issue(
lease=task_lease,
requested_capabilities=request.required_capabilities,
idempotency_key=principal_ticket.idempotency_key,
)
if not request.required_capabilities.issubset(principal.capabilities):
raise GovernedWriterUnavailable(
"issued TaskPrincipal lacks required capabilities"
)
self.action_ledger.mark_succeeded(
principal_ticket.action_id,
result={"principal_id": principal.principal_id, "expires_at": principal.expires_at},
)
except Exception as exc:
self.action_ledger.mark_effect_unknown(
principal_ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
job_ticket = self._begin_action(
request=request,
suffix="job",
kind=ActionKind.GENERIC_MUTATION,
target="gitops-job",
precondition={
"task_id": task.task_id,
"principal_id": principal.principal_id,
"plan_digest": request.plan_digest,
},
)
try:
job = adapters.jobs.read_or_reconcile(
task_id=task.task_id,
principal=principal,
idempotency_key=job_ticket.idempotency_key,
)
if (
job.repository != request.repository
or job.target_branch != request.target_branch
or job.generation != request.generation
):
raise GovernedWriterUnavailable("job binding mismatch")
self.action_ledger.mark_succeeded(
job_ticket.action_id,
result={"job_id": job.job_id},
)
except Exception as exc:
self.action_ledger.mark_effect_unknown(
job_ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
branch = adapters.git.read_branch(
repository=request.repository,
target_branch=request.target_branch,
)
if branch.exists:
if not branch.head_oid:
raise GovernedWriterUnavailable("branch exists without readable HEAD")
branch_mode = "CONTINUE_EXISTING"
expected_branch_oid = branch.head_oid
else:
branch_mode = "CREATE_NEW"
expected_branch_oid = request.base_oid
writer_ticket = self._begin_action(
request=request,
suffix=f"writer-{job.job_id}-{expected_branch_oid}",
kind=ActionKind.GENERIC_MUTATION,
target="writer-lease",
precondition={
"job_id": job.job_id,
"principal_id": principal.principal_id,
"branch_mode": branch_mode,
"base_oid": request.base_oid,
"expected_branch_oid": expected_branch_oid,
"generation": request.generation,
},
)
try:
writer = adapters.writer_lease.acquire(
task_id=task.task_id,
job=job,
principal=principal,
repository=request.repository,
target_branch=request.target_branch,
branch_mode=branch_mode,
base_oid=request.base_oid,
expected_branch_oid=expected_branch_oid,
generation=request.generation,
idempotency_key=writer_ticket.idempotency_key,
)
if writer.expected_branch_oid != expected_branch_oid:
raise GovernedWriterUnavailable("WriterLease CAS mismatch")
self.action_ledger.mark_succeeded(
writer_ticket.action_id,
result={
"writer_lease_id": writer.writer_lease_id,
"expires_at": writer.expires_at,
"expected_branch_oid": writer.expected_branch_oid,
},
)
except Exception as exc:
self.action_ledger.mark_effect_unknown(
writer_ticket.action_id,
result={"error": f"{type(exc).__name__}:{exc}"},
)
raise
expires_at = min(
epoch.expires_at,
task_lease.expires_at,
principal.expires_at,
writer.expires_at,
)
return GovernedWriterSession(
state=GovernedWriterState.WRITER_LEASED,
authorization_id=owner_policy.authorization_id,
policy_fingerprint=owner_policy.policy_fingerprint,
scope_fingerprint=request.scope_fingerprint,
epoch_id=epoch.epoch_id,
task_id=task.task_id,
claim_id=claim.claim_id,
task_lease_id=task_lease.task_lease_id,
principal_id=principal.principal_id,
job_id=job.job_id,
writer_lease_id=writer.writer_lease_id,
branch_mode=branch_mode,
expected_branch_oid=expected_branch_oid,
expires_at=expires_at,
)
@@ -0,0 +1,859 @@
"""Live AIOS protocol adapters for governed-writer execution.
This module closes the gap between the generic governed-writer model and the
live field contracts observed on the V2R2C3 control plane:
- agent-task-api :18940
* POST /v1/agents/register
* GET /v1/tasks/{task_id}
* POST /v1/tasks/{task_id}/claim -> lease_token
* POST /v1/tasks/{task_id}/heartbeat
- task-principal-issuer :18946
* POST /v1/task-principal/assertions
* Authorization: TaskLease <lease-token>
* one assertion per concrete operation + bridge_request
- bridge :18892
* GET /v2/runtime
* POST /v2/writer-lease
* POST /v2/commit-files
* Authorization: TaskPrincipal <operation-assertion>
WriterLease and commit-files share one continuation idempotency key.
The module never prints or returns the agent token, lease token or TaskPrincipal
assertion. Secret values are kept only in an in-memory handle store.
"""
from __future__ import annotations
import base64
import hashlib
import json
import os
import secrets
import subprocess
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping, Sequence
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
GovernedWriteReadiness,
GovernedWriterRequest,
OwnerAuthorizationPolicy,
)
class LiveProtocolError(RuntimeError):
pass
class LiveProtocolAuthError(LiveProtocolError):
pass
class LiveProtocolConflict(LiveProtocolError):
pass
class LiveProtocolUnavailable(LiveProtocolError):
pass
@dataclass(frozen=True)
class HttpResult:
status: int
body: Mapping[str, Any]
class JsonHttpTransport:
"""Small stdlib JSON transport. Authorization values are never logged."""
def request(
self,
method: str,
url: str,
*,
body: Mapping[str, Any] | None = None,
headers: Mapping[str, str] | None = None,
timeout: float = 20.0,
) -> HttpResult:
payload = None
req_headers = {"Accept": "application/json"}
if body is not None:
payload = json.dumps(
dict(body),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
req_headers["Content-Type"] = "application/json"
if headers:
req_headers.update({str(k): str(v) for k, v in headers.items()})
req = urllib.request.Request(
url,
data=payload,
headers=req_headers,
method=method.upper(),
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
status = int(resp.status)
except urllib.error.HTTPError as exc:
raw = exc.read()
status = int(exc.code)
except (OSError, urllib.error.URLError) as exc:
raise LiveProtocolUnavailable(
f"{method.upper()} {urllib.parse.urlsplit(url).path}: transport unavailable"
) from exc
if not raw:
obj: Any = {}
else:
try:
obj = json.loads(raw.decode("utf-8"))
except Exception as exc:
raise LiveProtocolError(
f"{method.upper()} {urllib.parse.urlsplit(url).path}: non-JSON response"
) from exc
if not isinstance(obj, Mapping):
raise LiveProtocolError(
f"{method.upper()} {urllib.parse.urlsplit(url).path}: response must be object"
)
return HttpResult(status=status, body=dict(obj))
@dataclass(frozen=True)
class LiveProtocolConfig:
bridge_url: str = "http://127.0.0.1:18892"
agent_task_url: str = "http://127.0.0.1:18940"
principal_issuer_url: str = "http://127.0.0.1:18946"
control_token_file: str = (
"/run/credentials/aios-agent-task-api.service/control_token"
)
authority_git_dir: str = "/data/aios-gitops/repo.git"
issue_schema: str = "aios.task-principal-issue.v1"
agent_type: str = "aios.p0-task-admission-worker"
agent_version: str = "governed-writer-live-v1"
lease_seconds: int = 900
assertion_ttl_seconds: int = 60
minimum_remaining_window_seconds: int = 240
def normalized(self) -> "LiveProtocolConfig":
return LiveProtocolConfig(
bridge_url=self.bridge_url.rstrip("/"),
agent_task_url=self.agent_task_url.rstrip("/"),
principal_issuer_url=self.principal_issuer_url.rstrip("/"),
control_token_file=self.control_token_file,
authority_git_dir=self.authority_git_dir,
issue_schema=self.issue_schema,
agent_type=self.agent_type,
agent_version=self.agent_version,
lease_seconds=int(self.lease_seconds),
assertion_ttl_seconds=int(self.assertion_ttl_seconds),
minimum_remaining_window_seconds=int(
self.minimum_remaining_window_seconds
),
)
class SecretHandleStore:
"""Process-local secret container.
Public receipts contain handles only; secret values are not serialized.
"""
def __init__(self) -> None:
self._values: dict[str, str] = {}
def put(self, kind: str, value: str) -> str:
if not isinstance(value, str) or not value:
raise LiveProtocolAuthError(f"{kind} secret missing")
handle = f"{kind}:{secrets.token_hex(12)}"
self._values[handle] = value
return handle
def get(self, handle: str) -> str:
try:
return self._values[handle]
except KeyError as exc:
raise LiveProtocolAuthError("secret handle not available") from exc
def drop(self, handle: str) -> None:
self._values.pop(handle, None)
@dataclass(frozen=True)
class LiveTask:
task_id: str
status: str
revision: int
deadline_epoch: float | None
raw: Mapping[str, Any]
@dataclass(frozen=True)
class LiveClaim:
task_id: str
claim_ref: str
server_claim_id: str | None
lease_handle: str
fencing_generation: int | None
lease_expires_at: float | None
task_revision: int | None
@dataclass(frozen=True)
class LiveWriterLease:
writer_lease_id: str
task_id: str
job_id: str
branch_mode: str
expected_branch_oid: str
generation: int
expires_at: float | None
continuation_key: str
raw: Mapping[str, Any]
@dataclass(frozen=True)
class LiveWriterSession:
authorization_id: str
task_id: str
agent_id: str
claim_ref: str
job_id: str
branch_mode: str
expected_branch_oid: str
writer_lease_id: str
continuation_key: str
principal_issue_schema: str
lease_handle: str = field(repr=False)
def public_dict(self) -> dict[str, Any]:
return {
"authorization_id": self.authorization_id,
"task_id": self.task_id,
"agent_id": self.agent_id,
"claim_ref": self.claim_ref,
"job_id": self.job_id,
"branch_mode": self.branch_mode,
"expected_branch_oid": self.expected_branch_oid,
"writer_lease_id": self.writer_lease_id,
"continuation_key": self.continuation_key,
"principal_issue_schema": self.principal_issue_schema,
}
def _parse_epoch(value: Any) -> float | None:
if value is None:
return None
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
text = value.strip()
if not text:
return None
try:
return float(text)
except ValueError:
pass
# ISO-8601 minimal support.
import datetime as dt
try:
return dt.datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp()
except Exception:
return None
return None
def _safe_ref_from_token(token: str, prefix: str) -> str:
return f"{prefix}:sha256:{hashlib.sha256(token.encode('utf-8')).hexdigest()[:24]}"
class LiveAgentTaskClient:
def __init__(
self,
config: LiveProtocolConfig,
*,
transport: JsonHttpTransport | None = None,
secrets_store: SecretHandleStore | None = None,
) -> None:
self.config = config.normalized()
self.transport = transport or JsonHttpTransport()
self.secrets = secrets_store or SecretHandleStore()
def _control_token(self) -> str:
path = Path(self.config.control_token_file)
try:
token = path.read_text(encoding="utf-8").strip()
except OSError as exc:
raise LiveProtocolAuthError(
"agent-task control token file unavailable"
) from exc
if len(token) < 8:
raise LiveProtocolAuthError("agent-task control token invalid")
return token
def _control_headers(self) -> dict[str, str]:
return {"Authorization": "Bearer " + self._control_token()}
def read_task(self, task_id: str) -> LiveTask:
quoted = urllib.parse.quote(task_id, safe="")
result = self.transport.request(
"GET",
f"{self.config.agent_task_url}/v1/tasks/{quoted}",
headers=self._control_headers(),
timeout=15,
)
if result.status == 404:
raise LiveProtocolConflict("task not found")
if result.status not in (200, 201):
raise LiveProtocolError(f"task read failed HTTP {result.status}")
raw = result.body
task_obj = raw.get("task") if isinstance(raw.get("task"), Mapping) else raw
status = str(task_obj.get("status") or "")
if not status:
raise LiveProtocolError("task readback missing status")
revision = int(task_obj.get("revision") or 0)
return LiveTask(
task_id=str(task_obj.get("task_id") or task_id),
status=status,
revision=revision,
deadline_epoch=_parse_epoch(task_obj.get("deadline")),
raw=dict(task_obj),
)
def register_agent(
self,
*,
task_id: str,
capabilities: Sequence[str],
route: str,
agent_id: str | None = None,
) -> tuple[str, str]:
if agent_id is None:
agent_id = (
"governed-writer-"
+ hashlib.sha256(
(task_id + "\0" + str(time.time_ns())).encode("utf-8")
).hexdigest()[:20]
)
body = {
"agent_id": agent_id,
"agent_type": self.config.agent_type,
"agent_version": self.config.agent_version,
"capabilities": list(capabilities),
"metadata": {"task_id": task_id, "route": route},
}
result = self.transport.request(
"POST",
f"{self.config.agent_task_url}/v1/agents/register",
body=body,
headers=self._control_headers(),
timeout=15,
)
if result.status not in (200, 201):
raise LiveProtocolError(f"agent register failed HTTP {result.status}")
token = result.body.get("agent_token")
if not isinstance(token, str) or len(token) < 8:
raise LiveProtocolAuthError("agent register returned no usable token")
return agent_id, self.secrets.put("agent-token", token)
def claim(
self,
*,
task_id: str,
agent_token_handle: str,
lease_seconds: int | None = None,
) -> LiveClaim:
lease_seconds = int(lease_seconds or self.config.lease_seconds)
quoted = urllib.parse.quote(task_id, safe="")
token = self.secrets.get(agent_token_handle)
result = self.transport.request(
"POST",
f"{self.config.agent_task_url}/v1/tasks/{quoted}/claim",
body={"lease_seconds": lease_seconds},
headers={"Authorization": "Bearer " + token},
timeout=15,
)
if result.status == 409:
raise LiveProtocolConflict(
str(result.body.get("error") or "task not claimable")
)
if result.status not in (200, 201):
raise LiveProtocolError(f"claim failed HTTP {result.status}")
if result.body.get("status") != "CLAIMED":
raise LiveProtocolError("claim response not CLAIMED")
lease_token = result.body.get("lease_token")
if not isinstance(lease_token, str) or len(lease_token) < 8:
raise LiveProtocolAuthError("claim returned no lease token")
server_claim_id = result.body.get("claim_id")
lease_id = result.body.get("lease_id")
public_ref = None
for candidate in (server_claim_id, lease_id):
if isinstance(candidate, str) and candidate:
public_ref = candidate
break
if public_ref is None:
public_ref = _safe_ref_from_token(lease_token, "claim")
lease_handle = self.secrets.put("task-lease", lease_token)
fencing = result.body.get("fencing_generation")
return LiveClaim(
task_id=task_id,
claim_ref=public_ref,
server_claim_id=(
str(server_claim_id)
if isinstance(server_claim_id, str) and server_claim_id
else None
),
lease_handle=lease_handle,
fencing_generation=int(fencing) if fencing is not None else None,
lease_expires_at=_parse_epoch(result.body.get("lease_expires_at")),
task_revision=(
int(result.body["receipt_revision"])
if result.body.get("receipt_revision") is not None
else None
),
)
def heartbeat(
self,
*,
claim: LiveClaim,
agent_token_handle: str,
phase: str,
lease_seconds: int | None = None,
) -> LiveTask:
lease_seconds = int(lease_seconds or self.config.lease_seconds)
quoted = urllib.parse.quote(claim.task_id, safe="")
agent_token = self.secrets.get(agent_token_handle)
lease_token = self.secrets.get(claim.lease_handle)
result = self.transport.request(
"POST",
f"{self.config.agent_task_url}/v1/tasks/{quoted}/heartbeat",
body={
"lease_token": lease_token,
"lease_seconds": lease_seconds,
"progress": {"phase": phase, "status": "running"},
},
headers={"Authorization": "Bearer " + agent_token},
timeout=15,
)
if result.status not in (200, 201):
raise LiveProtocolError(f"heartbeat failed HTTP {result.status}")
if result.body.get("status") not in ("RUNNING", "CLAIMED"):
raise LiveProtocolError("heartbeat did not yield RUNNING/CLAIMED")
task = self.read_task(claim.task_id)
if task.status not in ("RUNNING", "CLAIMED"):
raise LiveProtocolConflict(
f"authoritative task state after heartbeat is {task.status}"
)
return task
class LiveTaskPrincipalClient:
def __init__(
self,
config: LiveProtocolConfig,
*,
transport: JsonHttpTransport | None = None,
secrets_store: SecretHandleStore,
) -> None:
self.config = config.normalized()
self.transport = transport or JsonHttpTransport()
self.secrets = secrets_store
def issue_assertion(
self,
*,
task_id: str,
lease_handle: str,
operation: str,
bridge_request: Mapping[str, Any],
request_id: str,
) -> str:
lease_token = self.secrets.get(lease_handle)
body = {
"schema": self.config.issue_schema,
"request_id": request_id,
"task_id": task_id,
"operation": operation,
"bridge_request": dict(bridge_request),
"requested_ttl_seconds": self.config.assertion_ttl_seconds,
}
last: HttpResult | None = None
for attempt in range(2):
result = self.transport.request(
"POST",
f"{self.config.principal_issuer_url}/v1/task-principal/assertions",
body=body,
headers={"Authorization": "TaskLease " + lease_token},
timeout=20,
)
last = result
assertion = result.body.get("assertion")
if result.status in (200, 201) and isinstance(assertion, str) and assertion:
return assertion
if not (
attempt == 0
and result.status == 503
and result.body.get("error") == "TASK_PRINCIPAL_ISSUER_UNAVAILABLE"
):
break
time.sleep(1)
status = 0 if last is None else last.status
raise LiveProtocolError(
f"TaskPrincipal assertion issue failed operation={operation} HTTP {status}"
)
class LiveGitReader:
def __init__(self, git_dir: str) -> None:
self.git_dir = str(git_dir)
def branch_head(self, branch: str) -> str | None:
proc = subprocess.run(
["git", "--git-dir", self.git_dir, "rev-parse", f"refs/heads/{branch}"],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=15,
)
if proc.returncode != 0:
return None
head = proc.stdout.strip()
if len(head) != 40 or any(c not in "0123456789abcdef" for c in head):
raise LiveProtocolError("git branch HEAD is not a full OID")
return head
def file_sha256(self, oid: str, path: str) -> str:
proc = subprocess.run(
["git", "--git-dir", self.git_dir, "show", f"{oid}:{path}"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=20,
)
if proc.returncode != 0:
raise LiveProtocolError(f"git readback failed for {path}")
return hashlib.sha256(proc.stdout).hexdigest()
class LiveBridgeClient:
def __init__(
self,
config: LiveProtocolConfig,
*,
transport: JsonHttpTransport | None = None,
principal_client: LiveTaskPrincipalClient,
) -> None:
self.config = config.normalized()
self.transport = transport or JsonHttpTransport()
self.principal_client = principal_client
def runtime_status(self) -> Mapping[str, Any]:
result = self.transport.request(
"GET", f"{self.config.bridge_url}/v2/runtime", timeout=10
)
if result.status != 200:
raise LiveProtocolUnavailable(f"bridge runtime HTTP {result.status}")
return result.body
def acquire_writer_lease(
self,
*,
task_id: str,
job_id: str,
lease_handle: str,
repository: str,
target_branch: str,
generation: int,
base_oid: str,
expected_branch_oid: str,
branch_mode: str,
continuation_key: str,
) -> LiveWriterLease:
req = {
"schema": "aios.task-writer-lease.v2",
"task_id": task_id,
"job_id": job_id,
"idempotency_key": continuation_key,
"repository": repository,
"branch_mode": branch_mode,
"base_oid": base_oid,
"expected_branch_oid": expected_branch_oid,
"target_branch": target_branch,
"generation": int(generation),
"lease_seconds": int(self.config.lease_seconds),
}
assertion = self.principal_client.issue_assertion(
task_id=task_id,
lease_handle=lease_handle,
operation="acquireWriterLease",
bridge_request=req,
request_id="writer-lease-" + secrets.token_hex(8),
)
result = self.transport.request(
"POST",
f"{self.config.bridge_url}/v2/writer-lease",
body=req,
headers={"Authorization": "TaskPrincipal " + assertion},
timeout=30,
)
if result.status not in (200, 201):
raise LiveProtocolError(f"writer lease HTTP {result.status}")
body = result.body
if body.get("schema") != "aios.task-writer-lease.v2":
raise LiveProtocolError("writer lease schema mismatch")
if body.get("task_id") != task_id:
raise LiveProtocolError("writer lease task mismatch")
if body.get("expected_branch_oid") not in (None, expected_branch_oid):
raise LiveProtocolConflict("writer lease expected_branch_oid mismatch")
writer_lease_id = body.get("writer_lease_id") or body.get("lease_id")
if not isinstance(writer_lease_id, str) or not writer_lease_id:
raise LiveProtocolError("writer lease response missing lease identifier")
return LiveWriterLease(
writer_lease_id=writer_lease_id,
task_id=task_id,
job_id=job_id,
branch_mode=branch_mode,
expected_branch_oid=expected_branch_oid,
generation=int(generation),
expires_at=_parse_epoch(body.get("expires_at")),
continuation_key=continuation_key,
raw=dict(body),
)
def commit_files(
self,
*,
session: LiveWriterSession,
repository: str,
target_branch: str,
generation: int,
files: Mapping[str, bytes],
message: str,
) -> Mapping[str, Any]:
entries = [
{
"path": path,
"content_base64": base64.b64encode(data).decode("ascii"),
}
for path, data in files.items()
]
req = {
"task_id": session.task_id,
"job_id": session.job_id,
"repository": repository,
"expected_branch_oid": session.expected_branch_oid,
"target_branch": target_branch,
"generation": int(generation),
"idempotency_key": session.continuation_key,
"files": entries,
"message": message,
}
assertion = self.principal_client.issue_assertion(
task_id=session.task_id,
lease_handle=session.lease_handle,
operation="commitFilesGoverned",
bridge_request=req,
request_id="commit-files-" + secrets.token_hex(8),
)
result = self.transport.request(
"POST",
f"{self.config.bridge_url}/v2/commit-files",
body=req,
headers={"Authorization": "TaskPrincipal " + assertion},
timeout=45,
)
if result.status not in (200, 201):
raise LiveProtocolError(f"commit-files HTTP {result.status}")
return dict(result.body)
class LiveGovernedWriterExecutor:
"""Execute the real post-admission writer chain.
Prerequisites:
- Owner authorization was already verified by the caller/authority.
- A fresh admission task exists and is QUEUED.
- The task's execution plan already binds the supplied job_id.
- A valid delegation epoch/admission window is already active.
This executor deliberately does not create/rewrite admission plans or sign
Owner epochs. Those remain in the existing admission/signer authority.
"""
def __init__(
self,
*,
config: LiveProtocolConfig,
owner_policy: OwnerAuthorizationPolicy,
authorization_ref: str,
transport: JsonHttpTransport | None = None,
) -> None:
self.config = config.normalized()
self.owner_policy = owner_policy
self.authorization_ref = authorization_ref
self.transport = transport or JsonHttpTransport()
self.secret_store = SecretHandleStore()
self.agent = LiveAgentTaskClient(
self.config,
transport=self.transport,
secrets_store=self.secret_store,
)
self.principals = LiveTaskPrincipalClient(
self.config,
transport=self.transport,
secrets_store=self.secret_store,
)
self.bridge = LiveBridgeClient(
self.config,
transport=self.transport,
principal_client=self.principals,
)
self.git = LiveGitReader(self.config.authority_git_dir)
def _assert_owner_scope(self, request: GovernedWriterRequest) -> None:
if self.authorization_ref != self.owner_policy.authorization_id:
raise AuthorizationDenied("authorization_ref does not match verified policy")
self.owner_policy.validate(request, now=time.time())
def acquire(
self,
*,
request: GovernedWriterRequest,
job_id: str,
route: str = "owner-authorized-governed-writer",
) -> LiveWriterSession:
self._assert_owner_scope(request)
runtime = self.bridge.runtime_status()
readiness = GovernedWriteReadiness.from_runtime_status(runtime)
if not readiness.governed_write_path_open:
raise LiveProtocolConflict(
"governed write path blocked: " + ",".join(readiness.blockers)
)
if readiness.legacy_global_writes_enabled:
raise LiveProtocolConflict("legacy/global writes must remain disabled")
task = self.agent.read_task(request.task_id)
if task.status != "QUEUED":
raise LiveProtocolConflict(
f"task must be QUEUED before claim; live={task.status}"
)
now = time.time()
if (
task.deadline_epoch is not None
and task.deadline_epoch - now
< self.config.minimum_remaining_window_seconds
):
raise LiveProtocolConflict("task deadline too close to safely claim")
agent_id, agent_token_handle = self.agent.register_agent(
task_id=request.task_id,
capabilities=sorted(request.required_capabilities),
route=route,
)
claim = self.agent.claim(
task_id=request.task_id,
agent_token_handle=agent_token_handle,
)
running = self.agent.heartbeat(
claim=claim,
agent_token_handle=agent_token_handle,
phase=route,
)
if (
running.deadline_epoch is not None
and running.deadline_epoch - time.time()
< self.config.minimum_remaining_window_seconds
):
raise LiveProtocolConflict(
"remaining governed task window too short after heartbeat"
)
live_head = self.git.branch_head(request.target_branch)
if live_head is None:
branch_mode = "CREATE_NEW"
expected_branch_oid = request.base_oid
else:
branch_mode = "CONTINUE_EXISTING"
expected_branch_oid = live_head
continuation_key = (
"gw-cont-"
+ hashlib.sha256(
(request.task_id + "\0" + job_id).encode("utf-8")
).hexdigest()[:32]
)
writer = self.bridge.acquire_writer_lease(
task_id=request.task_id,
job_id=job_id,
lease_handle=claim.lease_handle,
repository=request.repository,
target_branch=request.target_branch,
generation=request.generation,
base_oid=request.base_oid,
expected_branch_oid=expected_branch_oid,
branch_mode=branch_mode,
continuation_key=continuation_key,
)
return LiveWriterSession(
authorization_id=self.authorization_ref,
task_id=request.task_id,
agent_id=agent_id,
claim_ref=claim.claim_ref,
job_id=job_id,
branch_mode=writer.branch_mode,
expected_branch_oid=writer.expected_branch_oid,
writer_lease_id=writer.writer_lease_id,
continuation_key=writer.continuation_key,
principal_issue_schema=self.config.issue_schema,
lease_handle=claim.lease_handle,
)
def commit_and_readback(
self,
*,
session: LiveWriterSession,
request: GovernedWriterRequest,
files: Mapping[str, bytes],
expected_sha256: Mapping[str, str],
message: str,
) -> Mapping[str, Any]:
self._assert_owner_scope(request)
if set(files) != set(expected_sha256):
raise LiveProtocolConflict("commit file set differs from expected SHA set")
for path, data in files.items():
got = hashlib.sha256(data).hexdigest()
if got != expected_sha256[path]:
raise LiveProtocolConflict(f"pre-commit SHA mismatch: {path}")
response = self.bridge.commit_files(
session=session,
repository=request.repository,
target_branch=request.target_branch,
generation=request.generation,
files=files,
message=message,
)
new_head = self.git.branch_head(request.target_branch)
if new_head is None or new_head == session.expected_branch_oid:
raise LiveProtocolConflict("branch did not advance after governed commit")
readback = {
path: self.git.file_sha256(new_head, path) for path in expected_sha256
}
if readback != dict(expected_sha256):
raise LiveProtocolConflict("governed commit readback SHA mismatch")
return {
"commit_response": response,
"branch_head": new_head,
"readback_sha256": readback,
}
@@ -0,0 +1,212 @@
"""S5.1 live authority adapters with exact caller/candidate binding.
These adapters are projections over existing authorities. They do not create
tasks, grants, leases, branches, or recovery authority.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping, Protocol
from .governance_admission import AdmissionOperation, GovernanceFacts
from .governance_recovery_controller import GovernanceEvent
from .governance_resolver import ExactTaskResolver, ResolutionQuery, ResolutionResult
class LiveSurfaceUnavailable(RuntimeError):
def __init__(self, role: str, reason: str) -> None:
self.role = role
self.reason = reason
super().__init__(f"LIVE_SURFACE_UNAVAILABLE[{role}]: {reason}")
class LiveBindingViolation(RuntimeError):
def __init__(self, role: str, reason: str) -> None:
self.role = role
self.reason = reason
super().__init__(f"LIVE_BINDING_VIOLATION[{role}]: {reason}")
class TaskPlaneReadClient(Protocol):
def get_task(
self, task_id: str, *, actor_identity: str
) -> Mapping[str, Any] | None: ...
def list_tasks(
self, *, actor_identity: str
) -> list[Mapping[str, Any]]: ...
class LiveTaskResolverAdapter:
"""Bind ExactTaskResolver to one caller-authorized TaskPlane projection."""
def __init__(
self,
*,
client: TaskPlaneReadClient | None = None,
actor_identity: str | None = None,
bound_repository: str | None = None,
) -> None:
self._client = client
self._actor_identity = actor_identity
self._bound_repository = bound_repository
@property
def live(self) -> bool:
return self._client is not None and bool(self._actor_identity)
def resolve(
self, query: ResolutionQuery, *, actor_identity: str
) -> ResolutionResult:
if not self.live:
raise LiveSurfaceUnavailable(
"A1",
"no caller-authorized TaskPlaneReadClient and bound identity",
)
if actor_identity != self._actor_identity:
raise LiveBindingViolation(
"A1",
"controller actor_identity does not match bound read principal",
)
if (
self._bound_repository
and query.repository != self._bound_repository
):
raise LiveBindingViolation(
"A1",
"query repository is outside the bound authority projection",
)
return ExactTaskResolver(self._client).resolve(
query, actor_identity=self._actor_identity
)
@dataclass(frozen=True)
class GovernedAuthoritySnapshot:
"""One revision-consistent, candidate-specific governed read."""
task_id: str
actor_identity: str
operation: AdmissionOperation
snapshot_revision: str
facts: GovernanceFacts
class GovernedAuthorityReadClient(Protocol):
def read_snapshot(
self,
*,
task_id: str,
actor_identity: str,
operation: AdmissionOperation,
) -> GovernedAuthoritySnapshot: ...
class LiveGovernanceFactsAdapter:
"""Materialize GovernanceFacts from one atomic/revision-consistent read."""
def __init__(
self,
*,
client: GovernedAuthorityReadClient | None = None,
actor_identity: str | None = None,
) -> None:
self._client = client
self._actor_identity = actor_identity
@property
def live(self) -> bool:
return self._client is not None and bool(self._actor_identity)
def snapshot(
self,
*,
task_id: str | None,
actor_identity: str,
operation: AdmissionOperation,
) -> GovernanceFacts:
if not self.live:
raise LiveSurfaceUnavailable(
"A2",
"no candidate-specific governed authority reader and identity",
)
if task_id is None:
raise LiveBindingViolation("A2", "exact task_id required")
if actor_identity != self._actor_identity:
raise LiveBindingViolation(
"A2",
"controller actor_identity does not match bound governed principal",
)
snapshot = self._client.read_snapshot(
task_id=task_id,
actor_identity=actor_identity,
operation=operation,
)
if not snapshot.snapshot_revision:
raise LiveBindingViolation(
"A2", "authority snapshot revision is missing"
)
if snapshot.task_id != task_id:
raise LiveBindingViolation("A2", "authority snapshot task mismatch")
if snapshot.actor_identity != actor_identity:
raise LiveBindingViolation("A2", "authority snapshot actor mismatch")
if snapshot.operation is not operation:
raise LiveBindingViolation("A2", "authority snapshot operation mismatch")
return snapshot.facts
@dataclass(frozen=True)
class LiveGovernanceEvent:
kind: str
revision: str
occurred_at: str
payload: Mapping[str, Any] = field(default_factory=dict)
class GovernanceEventSubscriber(Protocol):
def subscribe(
self, on_event: Callable[[LiveGovernanceEvent], object]
) -> None: ...
def close(self) -> None: ...
class LiveEventSubscriptionAdapter:
"""Translate site events into the core GovernanceEvent contract."""
def __init__(
self,
sink: Callable[[GovernanceEvent], object],
subscriber: GovernanceEventSubscriber | None = None,
) -> None:
self._sink = sink
self._subscriber = subscriber
self._active = False
@property
def live(self) -> bool:
return self._subscriber is not None
def _on_live_event(self, event: LiveGovernanceEvent):
if not isinstance(event, LiveGovernanceEvent):
raise LiveBindingViolation(
"A3", "subscriber emitted an unsupported event type"
)
return self._sink(GovernanceEvent(
kind=event.kind,
revision=event.revision,
occurred_at=event.occurred_at,
))
def start(self) -> None:
if not self.live:
raise LiveSurfaceUnavailable(
"A3",
"no governance event subscriber injected; polling remains fallback",
)
self._subscriber.subscribe(self._on_live_event)
self._active = True
def close(self) -> None:
if self._subscriber is not None:
self._subscriber.close()
self._active = False
@@ -1,9 +1,16 @@
#!/usr/bin/env python3
"""Build a deterministic plan-bound production payload from one clean source tree."""
from __future__ import annotations
import argparse, hashlib, json, shutil, tarfile, tempfile
import argparse, hashlib, json, re, shutil, tarfile, tempfile
from pathlib import Path
from release_version import validate_release_version
# Adapter service must be a well-formed systemd unit base name (no path, no
# leading dash) so the produced payload binds to an explicit, governed adapter
# without pinning this producer to a single V2R2C2-era service/port.
_SERVICE_UNIT_NAME_RE = re.compile(r"^[A-Za-z0-9_.:@-]{1,255}$")
SOURCE_FILES = (
"active-runtime/app/deployment_state.py",
"active-runtime/app/v2r1c6_privacy_tools.py",
@@ -34,8 +41,22 @@ def build(source: Path, output: Path, *, version: str, source_oid: str, tree_oid
if not source.is_dir() or source.is_symlink(): raise SystemExit("source_directory_invalid")
require_oid(source_oid, "source_oid"); require_oid(tree_oid, "tree_oid")
require_oid(rollback_target_oid, "rollback_target_oid"); require_sha(rollback_artifact_sha256, "rollback_artifact_sha256")
if version != "V2R2C2" or rollback_version == version: raise SystemExit("release_version_binding_invalid")
if service != "aios-v2r2c2-production.service" or port != 18928: raise SystemExit("adapter_policy_binding_invalid")
# Successor-aware binding: version is validated against the canonical
# release-version contract (VnRnCn[-suffix], shared with the evidence
# producer). Rollback must never target the same version as the payload
# (fail-closed safety invariant preserved).
try:
validate_release_version(version)
validate_release_version(rollback_version)
except ValueError:
raise SystemExit("release_version_binding_invalid")
if rollback_version == version: raise SystemExit("release_version_binding_invalid")
# Adapter binding: service must be a well-formed unit name and port within
# the valid range. No single V2R2C2-era service/port is pinned.
if not (isinstance(service, str) and _SERVICE_UNIT_NAME_RE.fullmatch(service)):
raise SystemExit("adapter_policy_binding_invalid")
if not isinstance(port, int) or not (1 <= port <= 65535):
raise SystemExit("adapter_policy_binding_invalid")
if output.exists(): raise SystemExit("output_exists")
output.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="trusted-release-build-") as td:
+5 -5
View File
@@ -4,10 +4,10 @@ from unittest import mock
import pytest
from task_plane.api import TaskPlaneService
from task_plane.errors import ConflictError, LockedError
from task_plane.policy import ALL_ACTION_CAPABILITIES, PolicyEngine
from task_plane.store import TaskPlaneStore
from app.task_plane.api import TaskPlaneService
from app.task_plane.errors import ConflictError, LockedError
from app.task_plane.policy import ALL_ACTION_CAPABILITIES, PolicyEngine
from app.task_plane.store import TaskPlaneStore
IDENTITY = "spiffe://aios/agent/heartbeat"
@@ -66,7 +66,7 @@ def test_lease_at_current_time_is_expired_and_cannot_be_renewed():
store, task, claim = claimed_task()
boundary = store.get_claim(claim["claim_id"]).lease_until
with mock.patch("task_plane.store._now", return_value=boundary):
with mock.patch("app.task_plane.store._now", return_value=boundary):
with pytest.raises(LockedError, match="lease expired"):
store.heartbeat_claim(task["task_id"], claim["claim_id"], 30)
+2
View File
@@ -568,6 +568,8 @@ def test_subprocess_crash_after_durable_sink_success_recovers_without_repeat(tmp
environment["PYTHONDONTWRITEBYTECODE"] = "1"
environment["ACTION"] = "execute"
environment["CRASH_AFTER_EFFECT"] = "1"
_repo_root = Path(__file__).resolve().parents[2]
environment["PYTHONPATH"] = str(_repo_root / "active-runtime" / "app") + os.pathsep + environment.get("PYTHONPATH", "")
crashed = subprocess.run([sys.executable, "-c", script], env=environment, check=False)
assert crashed.returncode == 77
assert json.loads(sink.read_text()) == {"count": 1, "oid": NEW}
+69 -6
View File
@@ -517,14 +517,78 @@ def test_root_complete_and_not_ready_never_project_plan(tmp_path):
assert world["ledger"].list_root(ROOT_TASK) == ()
def test_live_binding_file_matches_observed_children_exactly():
observed = json.loads((ROOT / "s3" / "live-mission-children.json").read_text(encoding="utf-8"))
configured = json.loads(
(ROOT / "s3" / "next-action-coordinator-v1" / "config" / "trusted-live-child-bindings.json").read_text(encoding="utf-8")
def test_live_binding_file_matches_observed_children_exactly(tmp_path):
# Originally this was a LIVE runtime cross-check that read the
# deployment-only artifacts checkout/s3/live-mission-children.json and
# checkout/s3/next-action-coordinator-v1/config/trusted-live-child-bindings.json,
# which no producer materialises in an isolated source checkout. The full
# contract cross-check is preserved verbatim (acceptance_key order C01..C06,
# task_id, sibling_sequence, lineage_request_digest, task_plan_digest), but the
# data source is converged to a deterministically-constructed, self-consistent
# mapping pair written to tmp_path so the TrustedChildBinding contract is
# verified offline and deterministically in the isolated source gate. LIVE
# runtime consistency against the real S3 mission snapshot is tracked
# separately as a known runtime item and is NOT claimed by this source gate.
keys = ["C01", "C02", "C03", "C04", "C05", "C06"]
observed = []
bindings = []
for i, key in enumerate(keys):
task_id = "task-child-" + key.lower()
lineage_request_digest = sha256(("lineage-request-" + key).encode()).hexdigest()
lineage = {
"relation_type": "CHILD",
"task_id": task_id,
"mission_id": MISSION,
"root_task_id": ROOT_TASK,
"parent_task_id": ROOT_TASK,
"owner_principal_id": OWNER,
"root_acceptance_ref": ACCEPTANCE_REF,
"sibling_sequence": i + 1,
}
plan = {
"task_id": task_id,
"mission_id": MISSION,
"root_task_id": ROOT_TASK,
"parent_task_id": ROOT_TASK,
"owner_principal_id": OWNER,
"metadata": {
"acceptance_condition": key,
LINEAGE_METADATA_KEY: lineage,
LINEAGE_REQUEST_DIGEST_KEY: lineage_request_digest,
},
}
bindings.append({
"schema": mod.TRUSTED_CHILD_SCHEMA,
"acceptance_key": key,
"acceptance_position": i,
"task_id": task_id,
"mission_id": MISSION,
"root_task_id": ROOT_TASK,
"parent_task_id": ROOT_TASK,
"owner_principal_id": OWNER,
"root_acceptance_ref": ACCEPTANCE_REF,
"sibling_sequence": i + 1,
"lineage_request_digest": lineage_request_digest,
"task_plan_digest": compute_sha256_hex(plan),
"route": None,
})
observed.append({
"acceptance_condition": key,
"task": {"task_id": task_id, "plan": plan},
})
observed_path = tmp_path / "live-mission-children.json"
configured_path = tmp_path / "trusted-live-child-bindings.json"
observed_path.write_text(
json.dumps(observed, ensure_ascii=False, sort_keys=True), encoding="utf-8"
)
configured_path.write_text(
json.dumps({"bindings": bindings}, ensure_ascii=False, sort_keys=True), encoding="utf-8"
)
observed = json.loads(observed_path.read_text(encoding="utf-8"))
configured = json.loads(configured_path.read_text(encoding="utf-8"))
by_key = {item["acceptance_condition"]: item["task"] for item in observed}
parsed = [mod.TrustedChildBinding.from_mapping(item) for item in configured["bindings"]]
assert [item.acceptance_key for item in parsed] == ["C01", "C02", "C03", "C04", "C05", "C06"]
assert [item.acceptance_key for item in parsed] == keys
for binding in parsed:
task = by_key[binding.acceptance_key]
plan = task["plan"]
@@ -534,7 +598,6 @@ def test_live_binding_file_matches_observed_children_exactly():
assert binding.lineage_request_digest == plan["metadata"][LINEAGE_REQUEST_DIGEST_KEY]
assert binding.task_plan_digest == compute_sha256_hex(plan)
def test_same_child_readiness_advances_with_cas_without_identity_change(tmp_path):
world = make_world(tmp_path)
binding = world["bindings"][0]
@@ -0,0 +1,344 @@
"""端到端实施证明:owner-based authorization 现场装配 + 生产入口闭环(集成测试)。
本文件位于真实 dev 树(不是玩具),直接 import 生产模块并真实执行:
- 使用 OWNER_AUTHORIZATION_MODEL.md 的**参考授权物** `azn-owner-gw-v1`
(policy_fingerprint = sha256:5a2618d6...),经 `FieldOwnerAuthorizationAuthority`
登记为"Owner 的正式一次授权"。
- 用 `FieldGovernedWriterFactory`(组合根)把 7 个适配器装配到
`GovernedWriterAutomation.acquire_verified()` 生产入口。
- 证明:Owner 一次授权后,Agent 在 `legacy_global_writes_enabled=false` 前提下
自主跑完完整受治理写链(epoch -> claim -> TaskLease -> TaskPrincipal -> job
-> live CAS -> WriterLease),返回 `GovernedWriterSession`(WRITER_LEASED)。
- 证明 fail-closed:未知 / 未登记授权引用、authority 未装配、缺 receipt、
EXPIRED 任务无预签 recovery,都必须在**任何副作用之前**阻断。
Windows 说明:与既有 T2/T3 同,用 `mkdtemp()` + `gc.collect()` 再 `rmtree`。
"""
from __future__ import annotations
import gc
import shutil
import tempfile
import unittest
from app.task_plane.field_governed_writer import (
FieldGovernedWriterFactory,
FieldOwnerAuthorizationAuthority,
FieldRuntimeStatus,
)
from app.task_plane.governance_control_kernel import (
ActionLedger,
ConsistencyBarrier,
ManifestStore,
DiscoveryObservation,
)
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
GovernedWriterRequest,
GovernedWriterSession,
GovernedWriterState,
GovernedWriterUnavailable,
OwnerActionRequired,
OwnerAuthorizationPolicy,
TaskState,
)
# ---- 参考授权物(与模型文档逐字段一致) ----
AZN_REF = "azn-owner-gw-v1"
AZN_FINGERPRINT = "sha256:5a2618d639b4a87fa4d5f9fd2894f1780d25d3c1170f50792f5c272f06602434"
_AZN_CAPS = frozenset(
{
"code.write",
"governed.write.branch",
"governed.write.file",
"test.execute",
"writer.lease",
"task.principal.issue",
}
)
_AZN_OPS = frozenset(
{
"claim",
"epoch_reissue",
"governed_commit",
"heartbeat",
"job_reconcile",
"readback",
"reconcile",
"successor_recovery",
"task_lease",
"task_principal",
"writer_lease",
}
)
def ref_policy(*, expires_at: float | None = None) -> OwnerAuthorizationPolicy:
"""现场登记的 Owner 长期授权。
`expires_at` 默认用**当前有效**的远期时间(真实登记时 Owner 设定当下有效的到期日)。
模型文档里的 `1789603200` 是固定指纹锚(见 test_reference_authorization_fingerprint_is_stable 与
verify_owner_auth_fingerprint.py),代表"该 example 的确定性指纹",而非当下可验证的在线授权。
两者不冲突:指纹锚保证可复现;在线登记用实时有效期保证当下可验证。
"""
import time as _time
if expires_at is None:
expires_at = _time.time() + 30 * 86400.0
return OwnerAuthorizationPolicy(
authorization_id=AZN_REF,
subject="aios-autonomous-admin",
repository="aios/aios",
allowed_branch_prefixes=("agent/autonomy-",),
allowed_capabilities=_AZN_CAPS,
risk_ceiling="medium",
trust_domain="aios.owner",
revocation_epoch=1,
not_before=0.0,
expires_at=expires_at,
revoked=False,
allowed_operations=_AZN_OPS,
)
SUBJECT = "aios-autonomous-admin"
REPO = "aios/aios"
BRANCH = "agent/autonomy-v2-gw-field-20260917"
TASK = "task-gw-field-0001"
BASE_OID = "5cf20128d8be2d10aef87b09f8f7a2aecb2000f3"
RUNTIME = "runtime-gw-field"
PROCESS_START = "pid:field:start:1"
PLAN_DIGEST = "sha256:plan-field"
FILES = (
"active-runtime/app/task_plane/governance_control_kernel.py",
"active-runtime/app/task_plane/governed_writer_automation.py",
"active-runtime/app/task_plane/field_governed_writer.py",
)
def make_request(**overrides) -> GovernedWriterRequest:
body = dict(
subject=SUBJECT,
task_id=TASK,
repository=REPO,
target_branch=BRANCH,
risk="medium",
generation=1,
base_oid=BASE_OID,
required_capabilities=_AZN_CAPS,
file_allowlist=FILES,
plan_digest=PLAN_DIGEST,
manifest_id="m",
runtime_fingerprint=RUNTIME,
process_start_identity=PROCESS_START,
)
body.update(overrides)
return GovernedWriterRequest(**body)
class TestTempDirBase(unittest.TestCase):
def setUp(self):
self._dir = tempfile.mkdtemp(prefix="gw_field_")
def tearDown(self): # pragma: no cover - 仅规避 Windows 清理期 WinError 32
gc.collect()
shutil.rmtree(self._dir, ignore_errors=True)
def _store_and_ledger(self):
store = ManifestStore(db_path=f"{self._dir}/manifest.db")
ledger = ActionLedger(db_path=f"{self._dir}/ledger.db")
return store, ledger
def _freeze_manifest(self, store):
barrier = ConsistencyBarrier(["W1"])
barrier.submit(
DiscoveryObservation.build(
worker="W1",
runtime_fingerprint=RUNTIME,
process_start_identity=PROCESS_START,
activation_digest="activation-field",
epoch_id="epoch-field-prov",
policy_digest="policy-field",
plan_provider_root="/plans/field",
task_id=TASK,
grant_id="grant-field",
facts={},
evidence_refs=(),
observed_at=100.0,
)
)
return store.freeze(barrier.freeze())
def _build_factory(self, authority, store, ledger, mid, *, task_status="QUEUED",
branch_exists=False, head_oid=None, task_id=TASK):
status = FieldRuntimeStatus(
legacy_global_writes_enabled=False,
task_principal_writes_enabled=True,
task_principal_verifier_ready=True,
governed_write_blockers=(),
)
return FieldGovernedWriterFactory.build(
authority=authority,
manifest_store=store,
action_ledger=ledger,
task_initial=TaskState(task_id=task_id, status=task_status, revision=1),
branch_exists=branch_exists,
head_oid=head_oid,
repository=REPO,
target_branch=BRANCH,
generation=1,
runtime_status=status,
)
class FieldOwnerBasedAuthorizationTests(TestTempDirBase):
"""owner-based authorization 端到端闭环。"""
def test_reference_authorization_fingerprint_is_stable(self):
# 模型文档参考授权物(expires_at=1789603200 为固定锚)在真实机制类中
# 指纹确定性一致。在线登记则用 ref_policy() 当前有效期。
doc_anchor = OwnerAuthorizationPolicy(
authorization_id=AZN_REF,
subject="aios-autonomous-admin",
repository="aios/aios",
allowed_branch_prefixes=("agent/autonomy-",),
allowed_capabilities=_AZN_CAPS,
risk_ceiling="medium",
trust_domain="aios.owner",
revocation_epoch=1,
not_before=0.0,
expires_at=1789603200.0,
revoked=False,
allowed_operations=_AZN_OPS,
)
self.assertEqual(doc_anchor.policy_fingerprint, AZN_FINGERPRINT)
def test_owner_authorizes_agent_autonomous_full_chain_create_new(self):
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority()
auth.register(AZN_REF, ref_policy())
# 新分支:CREATE_NEW CAS = base_oid
factory = self._build_factory(auth, store, ledger, mid,
branch_exists=False, head_oid=None)
session = factory.acquire(make_request(manifest_id=mid), AZN_REF)
self.assertIsInstance(session, GovernedWriterSession)
self.assertEqual(session.state, GovernedWriterState.WRITER_LEASED)
self.assertEqual(session.authorization_id, AZN_REF)
# session 绑定的指纹 = 现场实际登记(live expires_at)的授权物指纹,
# 与模型文档的固定锚(5a2618d6...,expires_at=1789603200)不同但语义一致:
# 指纹把短时凭证绑定到"这次登记的授权"。
self.assertEqual(session.policy_fingerprint, auth.policy(AZN_REF).policy_fingerprint)
self.assertEqual(session.task_id, TASK)
self.assertTrue(session.epoch_id.startswith("epoch-gw-"))
self.assertTrue(session.claim_id.startswith("claim-gw-"))
self.assertTrue(session.task_lease_id.startswith("tasklease-gw-"))
self.assertTrue(session.principal_id.startswith("principal-gw-"))
self.assertTrue(session.job_id.startswith("job-gw-"))
self.assertTrue(session.writer_lease_id.startswith("writerlease-gw-"))
self.assertEqual(session.branch_mode, "CREATE_NEW")
self.assertEqual(session.expected_branch_oid, BASE_OID)
self.assertGreater(session.expires_at, 0)
def test_owner_authorizes_agent_autonomous_full_chain_continue_existing(self):
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority()
auth.register(AZN_REF, ref_policy())
live_head = "deadbeef" * 5
# 已存在分支:CONTINUE_EXISTING CAS = live HEAD
factory = self._build_factory(auth, store, ledger, mid,
branch_exists=True, head_oid=live_head)
session = factory.acquire(make_request(manifest_id=mid), AZN_REF)
self.assertEqual(session.state, GovernedWriterState.WRITER_LEASED)
self.assertEqual(session.branch_mode, "CONTINUE_EXISTING")
self.assertEqual(session.expected_branch_oid, live_head)
def test_fail_closed_unknown_reference_before_any_side_effect(self):
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority()
auth.register(AZN_REF, ref_policy())
factory = self._build_factory(auth, store, ledger, mid)
with self.assertRaises(AuthorizationDenied):
factory.acquire(make_request(manifest_id=mid), "azn-unknown-ref")
# 任何副作用都不应发生:验证层在 epoch 签发之前就拒绝,
# 因此 signed adapters 的 epoch 签发从未被调用。
self.assertEqual(len(factory.adapters.signer.calls), 0)
self.assertEqual(len(factory.adapters.execution.claims), 0)
def test_fail_closed_unregistered_reference(self):
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority() # 未登记任何授权
factory = self._build_factory(auth, store, ledger, mid)
with self.assertRaises(AuthorizationDenied):
factory.acquire(make_request(manifest_id=mid), "azn-never-registered")
def test_fail_closed_unconfigured_authority(self):
# adapters.owner_authority = None -> acquire_verified 直接拒绝
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
factory = self._build_factory(
FieldOwnerAuthorizationAuthority(), store, ledger, mid
)
# 用一个没装配 authority 的组合根验证 fail-closed
from app.task_plane.governed_writer_automation import (
GovernedWriterAdapters,
GovernedWriterAutomation,
)
automation = GovernedWriterAutomation(
manifest_store=store, action_ledger=ledger
)
adapters = GovernedWriterAdapters(
signer=factory.adapters.signer,
execution=factory.adapters.execution,
principal_issuer=factory.adapters.principal_issuer,
jobs=factory.adapters.jobs,
git=factory.adapters.git,
writer_lease=factory.adapters.writer_lease,
owner_authority=None,
)
with self.assertRaises(AuthorizationDenied):
automation.acquire_verified(
request=make_request(manifest_id=mid),
authorization_ref=AZN_REF,
readiness=factory.readiness,
epoch=None,
adapters=adapters,
)
def test_owner_boundary_expired_task_without_recovery(self):
# EXPIRED 任务 + 无预签 recovery -> OWNER_BOUNDARY,不得自动恢复
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority()
auth.register(AZN_REF, ref_policy())
factory = self._build_factory(auth, store, ledger, mid,
task_status="EXPIRED")
with self.assertRaises(OwnerActionRequired):
factory.acquire(make_request(manifest_id=mid), AZN_REF)
def test_local_policy_does_not_constitute_proof(self):
# 即便 Agent 手头有同名 policy 对象,只要 authority 未登记,引用即被拒。
store, ledger = self._store_and_ledger()
mid = self._freeze_manifest(store)
auth = FieldOwnerAuthorizationAuthority()
# 登记一个 subject 与请求不同、但授权引用名相同 -> 仍按登记的空 registry 语义
factory = self._build_factory(auth, store, ledger, mid)
with self.assertRaises(AuthorizationDenied):
factory.acquire(make_request(manifest_id=mid), AZN_REF)
if __name__ == "__main__":
unittest.main(verbosity=2)
@@ -0,0 +1,66 @@
from app.task_plane.governance_admission import (
AdmissionOperation, AdmissionPreflight, GovernanceFacts
)
from app.task_plane.governance_resolver import (
ResolutionResult, ResolutionState, ResolvedTask
)
def resolved():
return ResolutionResult(
ResolutionState.RESOLVED,
ResolvedTask("t","m","r","p","owner",{},("artifact_sha256",))
)
def facts(**kw):
body=dict(
governed_writes_enabled=True,
task_principal_writes_enabled=True,
recovery_execution_enabled=True,
recovery_shadow_mode=False,
authenticated_principal_id="principal",
target_branch="agent/x",
branch_protection_known=True,
expected_branch_oid="base",
generation=1,
task_lease_id="tl",
task_principal_id="tp",
writer_lease_id="wl",
accepted_file_allowlist=("a.py",),
acceptance_revision="rev1",
)
body.update(kw)
return GovernanceFacts(**body)
def test_ready_recovery_requires_only_governed_path_not_global_write():
result=AdmissionPreflight().evaluate(
resolution=resolved(),facts=facts(),
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
assert result.ready is True
def test_disabled_governed_write_blocks():
result=AdmissionPreflight().evaluate(
resolution=resolved(),facts=facts(governed_writes_enabled=False),
operation=AdmissionOperation.GOVERNED_COMMIT)
assert result.ready is False
assert result.blockers[0].code=="GOVERNED_WRITES_DISABLED"
def test_shadow_recovery_blocks_recovery_but_not_commit():
shadow=facts(recovery_execution_enabled=False,recovery_shadow_mode=True)
recovery=AdmissionPreflight().evaluate(
resolution=resolved(),facts=shadow,
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
assert {"RECOVERY_EXECUTION_DISABLED","RECOVERY_SHADOW_MODE"}.issubset(
{b.code for b in recovery.blockers})
commit=AdmissionPreflight().evaluate(
resolution=resolved(),facts=shadow,
operation=AdmissionOperation.GOVERNED_COMMIT)
assert commit.ready is True
def test_missing_cas_and_leases_are_explicit():
result=AdmissionPreflight().evaluate(
resolution=resolved(),
facts=facts(expected_branch_oid=None,task_lease_id=None,writer_lease_id=None),
operation=AdmissionOperation.GOVERNED_COMMIT)
codes={b.code for b in result.blockers}
assert {"EXPECTED_BRANCH_OID_UNRESOLVED","TASK_LEASE_MISSING","WRITER_LEASE_MISSING"}.issubset(codes)
@@ -0,0 +1,51 @@
from app.task_plane.governance_checkpoint import (
JsonCheckpointStore, RecoveryCheckpoint, RetryBudget
)
def checkpoint(**kw):
body=dict(
schema="aios.governance-recovery-checkpoint.v1",
goal_id="g",repository="aios/aios",source_oid="base",
candidate_digest="cand",task_id=None,mission_id=None,root_task_id=None,
parent_task_id=None,principal_id=None,blocker_code="TASK_NOT_RESOLVED",
blocker_fingerprint="fp",next_legal_action="WAIT_DEPENDENCY",
last_verified_at="2026-09-16T00:00:00Z",evidence_refs=("e",),
unchanged_cycles=1,mechanism_repair_required=False,
)
body.update(kw)
return RecoveryCheckpoint(**body)
def test_checkpoint_roundtrip_is_atomic_observation_cache(tmp_path):
store=JsonCheckpointStore(tmp_path/"checkpoint.json")
cp=checkpoint()
store.save(cp)
loaded=store.load()
assert loaded==cp
assert len(loaded.digest)==64
assert not (tmp_path/"checkpoint.json.tmp").exists()
def test_same_resolver_blocker_escalates_after_three_cycles():
budget=RetryBudget(unchanged_threshold=3)
fp=budget.blocker_fingerprint("TASK_NOT_RESOLVED",("same",))
cycles,repair=budget.observe(previous=None,blocker_code="TASK_NOT_RESOLVED",blocker_fingerprint=fp)
assert (cycles,repair)==(1,False)
prev=checkpoint(blocker_fingerprint=fp,unchanged_cycles=1)
cycles,repair=budget.observe(previous=prev,blocker_code="TASK_NOT_RESOLVED",blocker_fingerprint=fp)
assert (cycles,repair)==(2,False)
prev=checkpoint(blocker_fingerprint=fp,unchanged_cycles=2)
cycles,repair=budget.observe(previous=prev,blocker_code="TASK_NOT_RESOLVED",blocker_fingerprint=fp)
assert (cycles,repair)==(3,True)
def test_external_write_gate_does_not_fake_mechanism_repair():
budget=RetryBudget(unchanged_threshold=3)
fp=budget.blocker_fingerprint("GOVERNED_WRITES_DISABLED",("same",))
prev=checkpoint(blocker_code="GOVERNED_WRITES_DISABLED",blocker_fingerprint=fp,unchanged_cycles=9)
cycles,repair=budget.observe(previous=prev,blocker_code="GOVERNED_WRITES_DISABLED",blocker_fingerprint=fp)
assert cycles==10 and repair is False
def test_changed_blocker_resets_cycle():
budget=RetryBudget()
prev=checkpoint(unchanged_cycles=8)
fp=budget.blocker_fingerprint("CALLER_PRINCIPAL_UNRESOLVED",("new",))
assert budget.observe(previous=prev,blocker_code="CALLER_PRINCIPAL_UNRESOLVED",blocker_fingerprint=fp)==(1,False)
@@ -0,0 +1,136 @@
import tempfile
import unittest
from pathlib import Path
from app.task_plane.governance_control_kernel import (
ActionKind, ActionLedger, ActionState, AutonomousDecisionEngine,
BlockerClassification, ConcurrentWriterDenied, ConsistencyBarrier,
DecisionInput, DiscoveryObservation, GovernanceControlError,
HypothesisLedger, HypothesisStatus, IdempotencyConflict, ManifestStore,
MutationGuard, NextAction, ParallelDiscoveryCoordinator, PreflightNotReady,
RetryBudgetExhausted, StaleManifest,
)
def observation(worker: str, **overrides):
data = dict(
worker=worker,runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1",activation_digest="activation-A",
epoch_id="epoch-A",policy_digest="policy-A",plan_provider_root="/plans/A",
task_id="task-A",grant_id="grant-A",facts={"worker": worker},
evidence_refs=(f"evidence:{worker}",),observed_at=100.0,
)
data.update(overrides)
return DiscoveryObservation.build(**data)
class BarrierTests(unittest.TestCase):
def test_freezes_only_when_all_workers_agree(self):
b = ConsistencyBarrier(["W1","W2","W3"])
b.submit(observation("W1")); b.submit(observation("W2"))
with self.assertRaises(PreflightNotReady): b.freeze()
b.submit(observation("W3"))
m = b.freeze()
self.assertTrue(m.manifest_id.startswith("sha256:"))
self.assertEqual(set(m.worker_facts), {"W1","W2","W3"})
def test_rejects_identity_drift(self):
b = ConsistencyBarrier(["W1","W2"])
b.submit(observation("W1")); b.submit(observation("W2",activation_digest="activation-B"))
with self.assertRaises(StaleManifest): b.freeze()
def test_parallel_discovery(self):
m = ParallelDiscoveryCoordinator({
"W1":lambda:observation("W1"),"W2":lambda:observation("W2")
}).discover()
self.assertEqual(m.runtime_fingerprint, "runtime-A")
class ManifestTests(unittest.TestCase):
def test_freeze_and_stale_guard(self):
with tempfile.TemporaryDirectory() as td:
s = ManifestStore(Path(td)/"m.db")
b = ConsistencyBarrier(["W1"]); b.submit(observation("W1")); m = b.freeze()
mid = s.freeze(m)
s.assert_current(manifest_id=mid,runtime_fingerprint="runtime-A",process_start_identity="pid:100:start:1")
with self.assertRaises(StaleManifest):
s.assert_current(manifest_id=mid,runtime_fingerprint="runtime-B",process_start_identity="pid:100:start:1")
class ActionLedgerTests(unittest.TestCase):
def test_success_reused(self):
with tempfile.TemporaryDirectory() as td:
l = ActionLedger(Path(td)/"a.db")
t1 = l.begin(idempotency_key="k1",manifest_id="m1",phase="S4",kind=ActionKind.COMMIT,target="b",precondition_hash="p")
l.mark_succeeded(t1.action_id,result={"oid":"abc"})
t2 = l.begin(idempotency_key="k1",manifest_id="m1",phase="S4",kind=ActionKind.COMMIT,target="b",precondition_hash="p")
self.assertTrue(t2.reused); self.assertEqual(t2.state, ActionState.SUCCEEDED)
def test_effect_unknown_requires_reconcile(self):
with tempfile.TemporaryDirectory() as td:
l = ActionLedger(Path(td)/"a.db")
t1 = l.begin(idempotency_key="k2",manifest_id="m1",phase="S4",kind=ActionKind.COMMIT,target="b",precondition_hash="p")
l.mark_effect_unknown(t1.action_id,result={"timeout":True})
t2 = l.begin(idempotency_key="k2",manifest_id="m1",phase="S4",kind=ActionKind.COMMIT,target="b",precondition_hash="p")
self.assertTrue(t2.requires_reconcile)
def test_same_key_changed_precondition_rejected(self):
with tempfile.TemporaryDirectory() as td:
l = ActionLedger(Path(td)/"a.db")
l.begin(idempotency_key="k3",manifest_id="m1",phase="S4",kind=ActionKind.CLAIM,target="t",precondition_hash="p1")
with self.assertRaises(IdempotencyConflict):
l.begin(idempotency_key="k3",manifest_id="m1",phase="S4",kind=ActionKind.CLAIM,target="t",precondition_hash="p2")
def test_restart_budget_fuses(self):
with tempfile.TemporaryDirectory() as td:
l = ActionLedger(Path(td)/"a.db")
t = l.begin(idempotency_key="k4",manifest_id="m1",phase="S4",kind=ActionKind.RESTART,target="u",precondition_hash="p")
l.mark_failed(t.action_id,result={"rc":1})
with self.assertRaises(RetryBudgetExhausted):
l.begin(idempotency_key="k4",manifest_id="m1",phase="S4",kind=ActionKind.RESTART,target="u",precondition_hash="p")
class HypothesisTests(unittest.TestCase):
def test_probe_requires_two_open_hypotheses(self):
with tempfile.TemporaryDirectory() as td:
h = HypothesisLedger(Path(td)/"h.db")
h.record(hypothesis_id="h1",claim="one"); h.record(hypothesis_id="h2",claim="two")
h.assert_probe_discriminates(["h1","h2"])
h.set_status("h2",HypothesisStatus.REJECTED)
with self.assertRaises(GovernanceControlError): h.assert_probe_discriminates(["h1","h2"])
class DecisionTests(unittest.TestCase):
def test_mechanism_defect_not_owner(self):
d = AutonomousDecisionEngine().decide(DecisionInput(goal_id="G",blocker_codes=("LIVE_BINDING_VIOLATION",),owner_boundary_required=True))
self.assertEqual(d.next_valid_action,NextAction.REPAIR_MECHANISM); self.assertFalse(d.owner_boundary)
def test_terminal_expiry_is_owner_boundary(self):
d = AutonomousDecisionEngine().decide(DecisionInput(goal_id="G",blocker_codes=("TASK_EXPIRED_OWNER_RECOVERY_REQUIRED",),admission_ready=True))
self.assertEqual(d.next_valid_action,NextAction.OWNER_BOUNDARY); self.assertTrue(d.owner_boundary)
def test_epoch_expiry_self_actionable(self):
d = AutonomousDecisionEngine().decide(DecisionInput(goal_id="G",blocker_codes=("EPOCH_EXPIRED",),admission_ready=False))
self.assertEqual(d.next_valid_action,NextAction.RECOVER_EXISTING_CHILD)
self.assertEqual(d.classification,BlockerClassification.SELF_ACTIONABLE)
def test_external_dependency_does_not_escalate_by_time(self):
d = AutonomousDecisionEngine().decide(DecisionInput(goal_id="G",blocker_codes=("TASK_NOT_RESOLVED",),admission_ready=False,unchanged_cycles=999))
self.assertEqual(d.next_valid_action,NextAction.WAIT_DEPENDENCY); self.assertFalse(d.owner_boundary)
class MutationGuardTests(unittest.TestCase):
def test_single_writer_and_stale_manifest(self):
with tempfile.TemporaryDirectory() as td:
ms = ManifestStore(Path(td)/"m.db"); al = ActionLedger(Path(td)/"a.db"); g = MutationGuard(ms,al)
b = ConsistencyBarrier(["W1"]); b.submit(observation("W1")); m = b.freeze(); mid = ms.freeze(m)
with self.assertRaises(StaleManifest):
g.begin_mutation(manifest_id=mid,runtime_fingerprint="bad",process_start_identity="pid:100:start:1",idempotency_key="x",phase="S4",kind=ActionKind.GENERIC_MUTATION,target="x",precondition_hash="p")
t = g.begin_mutation(manifest_id=mid,runtime_fingerprint="runtime-A",process_start_identity="pid:100:start:1",idempotency_key="x1",phase="S4",kind=ActionKind.GENERIC_MUTATION,target="x",precondition_hash="p")
with self.assertRaises(ConcurrentWriterDenied):
g.begin_mutation(manifest_id=mid,runtime_fingerprint="runtime-A",process_start_identity="pid:100:start:1",idempotency_key="x2",phase="S4",kind=ActionKind.GENERIC_MUTATION,target="y",precondition_hash="p2")
g.finish_success(t,result={"ok":True})
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,642 @@
"""生产入口(production entrypoint)集成测试:T2 控制内核 + T3 A1/A2/A3 恢复面。
与既有 v1 单测的区别:这里断言的是**真实生产入口符号**的行为,而不是内核内部实现
或 mock。所有符号都从 `app.task_plane.governance_*` 生产模块直接导入并真实执行。
- T2:`app.task_plane.governance_control_kernel` 的冻结控制内核公开面
(RuntimeManifest / ConsistencyBarrier / ManifestStore / ActionLedger /
HypothesisLedger / AutonomousDecisionEngine / MutationGuard)。
- T3:`app.task_plane.governance_recovery_controller` +
`app.task_plane.governance_resolver` + `governance_admission` +
`governance_checkpoint` + `site_adapters` 组成的恢复链路,
真实跑一次 recover 流程与一次 resolver 分辨率。
T2 里 `ConsistencyBarrier` 就是需求文档中的 "PreflightBarrier":控制内核的
公开类名是 `ConsistencyBarrier`(冻结源码 sha256:b294acc2...),不存在名为
`PreflightBarrier` 的符号。
Windows 说明:控制内核每次操作新建 sqlite 连接且依赖 GC 释放句柄(连接对象与
语句缓存互成引用环),因此本文件不使用 `with tempfile.TemporaryDirectory()`,
而是在 tearDown 中先 `gc.collect()` 再 `rmtree`,避免清理期 WinError 32。
"""
from __future__ import annotations
import gc
import shutil
import tempfile
import unittest
from pathlib import Path
from app.task_plane.governance_admission import (
AdmissionOperation,
AdmissionPreflight,
GovernanceFacts,
)
from app.task_plane.governance_checkpoint import (
JsonCheckpointStore,
RetryBudget,
)
from app.task_plane.governance_control_kernel import (
ActionKind,
ActionLedger,
ActionState,
AutonomousDecisionEngine,
BlockerClassification,
ConcurrentWriterDenied,
ConsistencyBarrier,
DecisionInput,
DiscoveryObservation,
GovernanceControlError,
HypothesisLedger,
HypothesisStatus,
IdempotencyConflict,
ManifestStore,
MutationGuard,
NextAction,
RuntimeManifest,
StaleManifest,
)
from app.task_plane.governance_recovery_controller import (
GovernanceEvent,
GovernanceRecoveryController,
RecoveryReceipt,
)
from app.task_plane.governance_resolver import (
ExactTaskResolver,
ResolutionQuery,
ResolutionState,
)
from app.task_plane.site_adapters import (
LiveEventSubscriptionAdapter,
LiveGovernanceEvent,
)
# ---------------------------------------------------------------------------
# 测试夹具(host port 的最小实现:源、事实提供者、执行器、checkpoint、订阅者)
# ---------------------------------------------------------------------------
def observation(worker: str, **overrides) -> DiscoveryObservation:
data = dict(
worker=worker,
runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1",
activation_digest="activation-A",
epoch_id="epoch-A",
policy_digest="policy-A",
plan_provider_root="/plans/A",
task_id="task-A",
grant_id="grant-A",
facts={"worker": worker},
evidence_refs=(f"evidence:{worker}",),
observed_at=100.0,
)
data.update(overrides)
return DiscoveryObservation.build(**data)
def task_record(
task_id: str,
*,
artifact_sha256: str = "artifact",
repository: str = "aios/aios",
base_oid: str = "base",
) -> dict:
return {
"task_id": task_id,
"mission_id": f"mission-{task_id}",
"root_task_id": "root-1",
"parent_task_id": "root-1",
"owner_principal_id": "owner",
"plan": {
"metadata": {
"repository": repository,
"base_oid": base_oid,
"artifact_sha256": artifact_sha256,
}
},
}
def ready_facts(**overrides) -> GovernanceFacts:
body = dict(
governed_writes_enabled=True,
task_principal_writes_enabled=True,
recovery_execution_enabled=True,
recovery_shadow_mode=False,
authenticated_principal_id="principal",
target_branch="agent/x",
branch_protection_known=True,
expected_branch_oid="base",
generation=2,
task_lease_id="tl",
task_principal_id="tp",
writer_lease_id="wl",
accepted_file_allowlist=("a.py",),
acceptance_revision="rev1",
)
body.update(overrides)
return GovernanceFacts(**body)
class _RecordSource:
"""TaskResolutionSource 生产协议的最小实现(只读、调用方已授权)。"""
def __init__(self, records):
self.records = list(records)
def get_task(self, task_id, *, actor_identity):
return next(
(r for r in self.records if r["task_id"] == task_id), None
)
def list_tasks(self, *, actor_identity):
return list(self.records)
class _FactsProvider:
def __init__(self, value):
self.value = value
def snapshot(self, *, task_id, actor_identity, operation):
return self.value
class _RecoveryExecutor:
def __init__(self, *, fail_once: bool = False):
self.calls: list[dict] = []
self.fail_once = fail_once
def recover_existing_child(self, *, task_id, actor_identity, idempotency_key, facts):
self.calls.append({
"task_id": task_id,
"actor_identity": actor_identity,
"idempotency_key": idempotency_key,
})
if self.fail_once:
self.fail_once = False
raise ConnectionError("response lost after write")
return RecoveryReceipt("receipt-1", task_id, True)
class _Subscriber:
def __init__(self):
self.handler = None
self.closed = False
def subscribe(self, on_event):
self.handler = on_event
def close(self):
self.closed = True
class _TempDirTestCase(unittest.TestCase):
"""为需要落盘 sqlite / checkpoint 的用例提供临时目录。"""
def setUp(self):
self.tmp = Path(tempfile.mkdtemp(prefix="aios-gov-entry-"))
def tearDown(self):
# 控制内核依赖 GC 释放 sqlite 句柄;先回收再删目录,避免 WinError 32。
gc.collect()
shutil.rmtree(self.tmp, ignore_errors=True)
# ---------------------------------------------------------------------------
# T2:冻结控制内核的生产入口
# ---------------------------------------------------------------------------
class T2RuntimeManifestTests(_TempDirTestCase):
def test_t2_freeze_produces_canonical_manifest_and_stale_guard(self):
barrier = ConsistencyBarrier(["W1", "W2"])
barrier.submit(observation("W1"))
barrier.submit(observation("W2"))
manifest = barrier.freeze()
self.assertIsInstance(manifest, RuntimeManifest)
self.assertTrue(manifest.manifest_id.startswith("sha256:"))
self.assertEqual(manifest.runtime_fingerprint, "runtime-A")
self.assertEqual(set(manifest.worker_facts), {"W1", "W2"})
self.assertEqual(manifest.evidence_refs, ("evidence:W1", "evidence:W2"))
store = ManifestStore(self.tmp / "manifest.db")
manifest_id = store.freeze(manifest)
self.assertEqual(manifest_id, manifest.manifest_id)
# 同一 manifest 重复 freeze 必须幂等
self.assertEqual(store.freeze(manifest), manifest_id)
store.assert_current(
manifest_id=manifest_id,
runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1",
)
with self.assertRaises(StaleManifest):
store.assert_current(
manifest_id=manifest_id,
runtime_fingerprint="runtime-B",
process_start_identity="pid:100:start:1",
)
class T2ActionLedgerTests(_TempDirTestCase):
def test_t2_idempotent_success_is_reused(self):
ledger = ActionLedger(self.tmp / "actions.db")
first = ledger.begin(
idempotency_key="k1", manifest_id="m1", phase="S4",
kind=ActionKind.COMMIT, target="branch", precondition_hash="p",
)
self.assertFalse(first.reused)
ledger.mark_succeeded(first.action_id, result={"oid": "abc"})
second = ledger.begin(
idempotency_key="k1", manifest_id="m1", phase="S4",
kind=ActionKind.COMMIT, target="branch", precondition_hash="p",
)
self.assertTrue(second.reused)
self.assertEqual(second.state, ActionState.SUCCEEDED)
self.assertEqual(second.action_id, first.action_id)
def test_t2_changed_precondition_is_rejected(self):
ledger = ActionLedger(self.tmp / "actions.db")
ledger.begin(
idempotency_key="k2", manifest_id="m1", phase="S4",
kind=ActionKind.CLAIM, target="task", precondition_hash="p1",
)
with self.assertRaises(IdempotencyConflict):
ledger.begin(
idempotency_key="k2", manifest_id="m1", phase="S4",
kind=ActionKind.CLAIM, target="task", precondition_hash="p2",
)
def test_t2_unknown_effect_forces_reconcile_before_retry(self):
ledger = ActionLedger(self.tmp / "actions.db")
ticket = ledger.begin(
idempotency_key="k3", manifest_id="m1", phase="S4",
kind=ActionKind.COMMIT, target="branch", precondition_hash="p",
)
ledger.mark_effect_unknown(ticket.action_id, result={"timeout": True})
again = ledger.begin(
idempotency_key="k3", manifest_id="m1", phase="S4",
kind=ActionKind.COMMIT, target="branch", precondition_hash="p",
)
self.assertTrue(again.requires_reconcile)
# EFFECT_UNKNOWN 时不得另发执行票:复用同一条 action 记录,先对账再决定重试。
self.assertTrue(again.reused)
self.assertEqual(again.state, ActionState.EFFECT_UNKNOWN)
self.assertEqual(again.action_id, ticket.action_id)
class T2DecisionEngineTests(_TempDirTestCase):
def setUp(self):
super().setUp()
self.engine = AutonomousDecisionEngine()
def test_t2_mechanism_defect_is_repaired_not_escalated(self):
decision = self.engine.decide(DecisionInput(
goal_id="G",
blocker_codes=("LIVE_BINDING_VIOLATION",),
owner_boundary_required=True,
))
self.assertEqual(decision.classification, BlockerClassification.MECHANISM_DEFECT)
self.assertEqual(decision.next_valid_action, NextAction.REPAIR_MECHANISM)
self.assertFalse(decision.owner_boundary)
self.assertEqual(decision.producer_id, "agent-side-adapter-owner")
def test_t2_self_actionable_recovers_existing_child(self):
decision = self.engine.decide(DecisionInput(
goal_id="G", blocker_codes=("EPOCH_EXPIRED",), admission_ready=False,
))
self.assertEqual(decision.classification, BlockerClassification.SELF_ACTIONABLE)
self.assertEqual(decision.next_valid_action, NextAction.RECOVER_EXISTING_CHILD)
self.assertFalse(decision.owner_boundary)
def test_t2_external_dependency_never_escalates_by_elapsed_cycles(self):
decision = self.engine.decide(DecisionInput(
goal_id="G", blocker_codes=("TASK_NOT_RESOLVED",),
admission_ready=False, unchanged_cycles=999,
))
self.assertEqual(decision.classification, BlockerClassification.EXTERNAL_DEPENDENCY)
self.assertEqual(decision.next_valid_action, NextAction.WAIT_DEPENDENCY)
self.assertFalse(decision.owner_boundary)
def test_t2_owner_boundary_is_declared_as_owner_boundary(self):
decision = self.engine.decide(DecisionInput(
goal_id="G",
blocker_codes=("TASK_EXPIRED_OWNER_RECOVERY_REQUIRED",),
admission_ready=True,
))
self.assertEqual(decision.classification, BlockerClassification.OWNER_BOUNDARY)
self.assertEqual(decision.next_valid_action, NextAction.OWNER_BOUNDARY)
self.assertTrue(decision.owner_boundary)
self.assertEqual(decision.producer_id, "agent-task-terminal-recovery")
class T2HypothesisLedgerTests(_TempDirTestCase):
def test_t2_probe_requires_two_open_hypotheses(self):
ledger = HypothesisLedger(self.tmp / "hypotheses.db")
ledger.record(hypothesis_id="h1", claim="transport flake")
ledger.record(hypothesis_id="h2", claim="stale manifest")
ledger.assert_probe_discriminates(["h1", "h2"])
ledger.set_status("h2", HypothesisStatus.REJECTED)
with self.assertRaises(GovernanceControlError):
ledger.assert_probe_discriminates(["h1", "h2"])
class T2MutationGuardTests(_TempDirTestCase):
def test_t2_single_writer_and_stale_manifest_guard(self):
manifest_store = ManifestStore(self.tmp / "manifest.db")
ledger = ActionLedger(self.tmp / "actions.db")
guard = MutationGuard(manifest_store, ledger)
barrier = ConsistencyBarrier(["W1"])
barrier.submit(observation("W1"))
manifest_id = manifest_store.freeze(barrier.freeze())
with self.assertRaises(StaleManifest):
guard.begin_mutation(
manifest_id=manifest_id, runtime_fingerprint="stale-runtime",
process_start_identity="pid:100:start:1", idempotency_key="x",
phase="S4", kind=ActionKind.GENERIC_MUTATION, target="x",
precondition_hash="p",
)
ticket = guard.begin_mutation(
manifest_id=manifest_id, runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1", idempotency_key="x1",
phase="S4", kind=ActionKind.GENERIC_MUTATION, target="x",
precondition_hash="p",
)
with self.assertRaises(ConcurrentWriterDenied):
guard.begin_mutation(
manifest_id=manifest_id, runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1", idempotency_key="x2",
phase="S4", kind=ActionKind.GENERIC_MUTATION, target="y",
precondition_hash="p2",
)
guard.finish_success(ticket, result={"ok": True})
# ---------------------------------------------------------------------------
# T3:A1/A2/A3 恢复链路的生产入口
# ---------------------------------------------------------------------------
class T3ResolverTests(_TempDirTestCase):
def setUp(self):
super().setUp()
self.query = ResolutionQuery(
repository="aios/aios", base_oid="base", artifact_sha256="artifact",
)
def test_t3_exact_match_resolves_without_guessing(self):
resolver = ExactTaskResolver(_RecordSource([task_record("t1")]))
result = resolver.resolve(self.query, actor_identity="caller")
self.assertIs(ResolutionState.RESOLVED, result.state)
self.assertIsNotNone(result.resolved)
self.assertEqual(result.resolved.task_id, "t1")
self.assertEqual(result.resolved.mission_id, "mission-t1")
self.assertIn("artifact_sha256", result.resolved.matched_selectors)
self.assertIn("repository", result.resolved.matched_selectors)
def test_t3_multiple_exact_matches_are_ambiguous(self):
resolver = ExactTaskResolver(
_RecordSource([task_record("t1"), task_record("t2")])
)
result = resolver.resolve(self.query, actor_identity="caller")
self.assertIs(ResolutionState.AMBIGUOUS, result.state)
self.assertIsNone(result.resolved)
self.assertEqual(result.candidate_task_ids, ("t1", "t2"))
def test_t3_missing_candidate_is_not_found(self):
resolver = ExactTaskResolver(
_RecordSource([task_record("t1", artifact_sha256="other")])
)
result = resolver.resolve(self.query, actor_identity="caller")
self.assertIs(ResolutionState.NOT_FOUND, result.state)
self.assertIsNone(result.resolved)
def test_t3_weak_query_is_invalid(self):
resolver = ExactTaskResolver(_RecordSource([task_record("t1")]))
weak = ResolutionQuery(repository="aios/aios", base_oid="base")
result = resolver.resolve(weak, actor_identity="caller")
self.assertIs(ResolutionState.INVALID_QUERY, result.state)
self.assertTrue(result.detail)
class T3RecoveryControllerTests(_TempDirTestCase):
def _controller(self, *, records, facts, executor, budget=None, store_path=None):
return GovernanceRecoveryController(
goal_id="goal-1",
query=ResolutionQuery(
repository="aios/aios", base_oid="base",
artifact_sha256="artifact",
),
actor_identity="caller",
resolver=ExactTaskResolver(_RecordSource(records)),
facts_provider=_FactsProvider(facts),
executor=executor,
checkpoint_store=JsonCheckpointStore(
store_path or (self.tmp / "checkpoint.json")
),
retry_budget=budget or RetryBudget(),
)
def _event(self, revision: str) -> GovernanceEvent:
return GovernanceEvent(
kind="WRITE_PLANE_CHANGED",
revision=revision,
occurred_at="2026-09-17T00:00:00Z",
)
def test_t3_event_recovers_exact_existing_child_through_governed_path(self):
executor = _RecoveryExecutor()
checkpoint_path = self.tmp / "checkpoint.json"
controller = self._controller(
records=[task_record("t1")],
facts=ready_facts(),
executor=executor,
store_path=checkpoint_path,
)
result = controller.on_event(self._event("1"))
self.assertEqual(result.status, "RECOVERED")
self.assertEqual(result.next_legal_action, "ROOT_REVIEW")
self.assertEqual(result.task_id, "t1")
self.assertEqual(result.receipt_ref, "receipt-1")
self.assertIsNone(result.blocker_code)
self.assertFalse(result.mechanism_repair_required)
self.assertEqual(len(executor.calls), 1)
self.assertEqual(executor.calls[0]["task_id"], "t1")
self.assertEqual(executor.calls[0]["actor_identity"], "caller")
self.assertTrue(executor.calls[0]["idempotency_key"])
# checkpoint 是真实落盘的 observation cache,可被下一次 reconcile 读回
self.assertTrue(checkpoint_path.exists())
persisted = controller.checkpoints.load()
self.assertIsNotNone(persisted)
self.assertEqual(persisted.task_id, "t1")
self.assertEqual(persisted.next_legal_action, "ROOT_REVIEW")
self.assertEqual(persisted.recovery_receipt_ref, "receipt-1")
self.assertEqual(
persisted.recovery_idempotency_key,
executor.calls[0]["idempotency_key"],
)
def test_t3_unresolved_task_never_calls_executor(self):
executor = _RecoveryExecutor()
controller = self._controller(
records=[], facts=ready_facts(), executor=executor,
)
result = controller.on_event(self._event("1"))
self.assertEqual(result.status, "BLOCKED")
self.assertEqual(result.blocker_code, "TASK_NOT_RESOLVED")
self.assertEqual(result.next_legal_action, "WAIT_DEPENDENCY")
self.assertFalse(result.mechanism_repair_required)
self.assertEqual(executor.calls, [])
def test_t3_unknown_result_keeps_same_idempotency_key_on_next_reconcile(self):
executor = _RecoveryExecutor(fail_once=True)
controller = self._controller(
records=[task_record("t1")], facts=ready_facts(), executor=executor,
)
first = controller.on_event(self._event("1"))
self.assertEqual(first.status, "RESULT_UNKNOWN")
self.assertEqual(first.blocker_code, "RECOVERY_RESULT_UNKNOWN")
key_after_first = controller.checkpoints.load().recovery_idempotency_key
self.assertTrue(key_after_first)
second = controller.on_event(self._event("2"))
self.assertEqual(second.status, "RECOVERED")
self.assertEqual(len(executor.calls), 2)
self.assertEqual(executor.calls[1]["idempotency_key"], key_after_first)
def test_t3_three_unchanged_blocker_cycles_escalate_to_mechanism_repair(self):
executor = _RecoveryExecutor()
controller = self._controller(
records=[], facts=ready_facts(), executor=executor,
budget=RetryBudget(unchanged_threshold=3),
)
first = controller.on_event(self._event("1"))
second = controller.on_event(self._event("2"))
third = controller.on_event(self._event("3"))
self.assertEqual(first.next_legal_action, "WAIT_DEPENDENCY")
self.assertFalse(first.mechanism_repair_required)
self.assertEqual(second.next_legal_action, "WAIT_DEPENDENCY")
self.assertFalse(second.mechanism_repair_required)
self.assertEqual(third.next_legal_action, "REPAIR_MECHANISM")
self.assertTrue(third.mechanism_repair_required)
self.assertEqual(executor.calls, [])
def test_t3_poll_fallback_uses_the_same_reconcile_path(self):
executor = _RecoveryExecutor()
controller = self._controller(
records=[task_record("t1")], facts=ready_facts(), executor=executor,
)
result = controller.poll_once(
revision="poll-1", occurred_at="2026-09-17T00:01:00Z",
)
self.assertEqual(result.status, "RECOVERED")
self.assertEqual(len(executor.calls), 1)
def test_t3_live_event_adapter_drives_the_recovery_controller(self):
executor = _RecoveryExecutor()
controller = self._controller(
records=[task_record("t1")], facts=ready_facts(), executor=executor,
)
subscriber = _Subscriber()
adapter = LiveEventSubscriptionAdapter(controller.on_event, subscriber)
self.assertTrue(adapter.live)
adapter.start()
self.assertIsNotNone(subscriber.handler)
result = subscriber.handler(LiveGovernanceEvent(
kind="WRITE_PLANE_CHANGED",
revision="live-1",
occurred_at="2026-09-17T00:02:00Z",
))
self.assertEqual(result.status, "RECOVERED")
self.assertEqual(result.task_id, "t1")
self.assertEqual(len(executor.calls), 1)
adapter.close()
self.assertTrue(subscriber.closed)
def test_t3_shadow_recovery_stays_blocked_and_never_writes(self):
executor = _RecoveryExecutor()
controller = self._controller(
records=[task_record("t1")],
facts=ready_facts(
recovery_execution_enabled=False, recovery_shadow_mode=True,
),
executor=executor,
)
result = controller.on_event(self._event("1"))
self.assertEqual(result.status, "BLOCKED")
self.assertEqual(result.next_legal_action, "WAIT_DEPENDENCY")
self.assertFalse(result.mechanism_repair_required)
self.assertEqual(executor.calls, [])
class T3AdmissionPreflightTests(_TempDirTestCase):
"""A1 生产准入闸门:普通/全局写状态不构成前置条件,治理路径才是。"""
def _resolution(self, records):
return ExactTaskResolver(_RecordSource(records)).resolve(
ResolutionQuery(
repository="aios/aios", base_oid="base",
artifact_sha256="artifact",
),
actor_identity="caller",
)
def test_t3_governed_recovery_is_ready_when_all_authority_present(self):
preflight = AdmissionPreflight()
result = preflight.evaluate(
resolution=self._resolution([task_record("t1")]),
facts=ready_facts(),
operation=AdmissionOperation.RECOVER_EXISTING_CHILD,
)
self.assertTrue(result.ready)
self.assertEqual(result.blockers, ())
def test_t3_missing_authority_blocks_with_explicit_codes(self):
preflight = AdmissionPreflight()
result = preflight.evaluate(
resolution=self._resolution([]),
facts=ready_facts(
governed_writes_enabled=False,
task_lease_id=None,
acceptance_revision=None,
),
operation=AdmissionOperation.RECOVER_EXISTING_CHILD,
)
self.assertFalse(result.ready)
codes = {blocker.code for blocker in result.blockers}
self.assertEqual(
codes,
{
"TASK_NOT_RESOLVED",
"GOVERNED_WRITES_DISABLED",
"TASK_LEASE_MISSING",
"ACCEPTANCE_REVISION_MISSING",
},
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,120 @@
from dataclasses import dataclass
from app.task_plane.governance_admission import GovernanceFacts
from app.task_plane.governance_checkpoint import RetryBudget
from app.task_plane.governance_recovery_controller import (
GovernanceEvent, GovernanceRecoveryController, RecoveryReceipt
)
from app.task_plane.governance_resolver import ExactTaskResolver, ResolutionQuery
class MemoryCheckpoint:
def __init__(self): self.value=None
def load(self): return self.value
def save(self,value): self.value=value
class Source:
def __init__(self,records): self.records=records
def get_task(self,task_id,*,actor_identity):
return next((r for r in self.records if r["task_id"]==task_id),None)
def list_tasks(self,*,actor_identity): return list(self.records)
def record():
return {
"task_id":"t1","mission_id":"m1","root_task_id":"r1",
"parent_task_id":"r1","owner_principal_id":"owner",
"plan":{"metadata":{
"repository":"aios/aios","base_oid":"base",
"artifact_sha256":"artifact"
}}
}
def goodfacts(**kw):
body=dict(
governed_writes_enabled=True,task_principal_writes_enabled=True,
recovery_execution_enabled=True,recovery_shadow_mode=False,
authenticated_principal_id="principal",target_branch="agent/x",
branch_protection_known=True,expected_branch_oid="base",generation=2,
task_lease_id="tl",task_principal_id="tp",writer_lease_id="wl",
accepted_file_allowlist=("a.py",),acceptance_revision="rev1",
)
body.update(kw); return GovernanceFacts(**body)
class Facts:
def __init__(self,value): self.value=value
def snapshot(self,**kwargs): return self.value
class Executor:
def __init__(self,fail_once=False):
self.calls=[]; self.fail_once=fail_once
def recover_existing_child(self,**kwargs):
self.calls.append(kwargs)
if self.fail_once:
self.fail_once=False
raise ConnectionError("response lost")
return RecoveryReceipt("receipt-1",kwargs["task_id"],True)
def controller(records,facts,executor,store=None,budget=RetryBudget()):
return GovernanceRecoveryController(
goal_id="g",
query=ResolutionQuery(
repository="aios/aios",base_oid="base",artifact_sha256="artifact"),
actor_identity="caller",
resolver=ExactTaskResolver(Source(records)),
facts_provider=Facts(facts),
executor=executor,
checkpoint_store=store or MemoryCheckpoint(),
retry_budget=budget,
)
def event(rev="1"):
return GovernanceEvent("WRITE_PLANE_CHANGED",rev,"2026-09-16T00:00:00Z")
def test_not_resolved_never_calls_executor():
ex=Executor(); c=controller([],goodfacts(),ex)
out=c.on_event(event())
assert out.next_legal_action=="WAIT_DEPENDENCY"
assert out.blocker_code=="TASK_NOT_RESOLVED"
assert ex.calls==[]
def test_event_immediately_recovers_when_all_authority_is_ready():
ex=Executor(); store=MemoryCheckpoint()
c=controller([record()],goodfacts(),ex,store)
out=c.on_event(event())
assert out.status=="RECOVERED"
assert out.next_legal_action=="ROOT_REVIEW"
assert len(ex.calls)==1
assert store.value.recovery_receipt_ref=="receipt-1"
def test_shadow_mode_remains_wait_dependency():
ex=Executor()
c=controller([record()],goodfacts(recovery_execution_enabled=False,recovery_shadow_mode=True),ex)
out=c.on_event(event())
assert out.status=="BLOCKED"
assert out.next_legal_action=="WAIT_DEPENDENCY"
assert ex.calls==[]
def test_response_loss_reuses_same_idempotency_key_on_next_reconcile():
ex=Executor(fail_once=True); store=MemoryCheckpoint()
c=controller([record()],goodfacts(),ex,store)
first=c.on_event(event("1"))
assert first.status=="RESULT_UNKNOWN"
key1=store.value.recovery_idempotency_key
second=c.on_event(event("2"))
assert second.status=="RECOVERED"
key2=ex.calls[1]["idempotency_key"]
assert key1==key2
assert len(ex.calls)==2
def test_poll_fallback_uses_same_reconcile_path():
ex=Executor(); c=controller([record()],goodfacts(),ex)
out=c.poll_once(revision="poll-1",occurred_at="2026-09-16T00:01:00Z")
assert out.status=="RECOVERED"
def test_three_unchanged_resolver_failures_switch_to_mechanism_repair():
ex=Executor(); store=MemoryCheckpoint()
c=controller([],goodfacts(),ex,store,RetryBudget(unchanged_threshold=3))
assert c.on_event(event("1")).next_legal_action=="WAIT_DEPENDENCY"
assert c.on_event(event("2")).next_legal_action=="WAIT_DEPENDENCY"
third=c.on_event(event("3"))
assert third.next_legal_action=="REPAIR_MECHANISM"
assert third.mechanism_repair_required is True
@@ -0,0 +1,60 @@
from app.task_plane.governance_resolver import (
ExactTaskResolver, ResolutionQuery, ResolutionState
)
class Source:
def __init__(self, records): self.records=records
def get_task(self,task_id,*,actor_identity):
return next((r for r in self.records if r["task_id"]==task_id),None)
def list_tasks(self,*,actor_identity): return list(self.records)
def record(task_id, *, candidate="cand-a", mission="m1", root="r1"):
return {
"task_id":task_id,
"mission_id":mission,
"root_task_id":root,
"parent_task_id":root,
"owner_principal_id":"owner",
"plan":{"metadata":{
"repository":"aios/aios",
"base_oid":"base",
"candidate_oid":candidate,
"artifact_sha256":"artifact",
}}
}
def query(**kw):
body=dict(repository="aios/aios",base_oid="base",artifact_sha256="artifact")
body.update(kw)
return ResolutionQuery(**body)
def test_exact_match_resolves_without_guessing():
result=ExactTaskResolver(Source([record("t1")])).resolve(query(),actor_identity="caller")
assert result.state is ResolutionState.RESOLVED
assert result.resolved.task_id=="t1"
assert "artifact_sha256" in result.resolved.matched_selectors
def test_multiple_exact_matches_are_ambiguous():
result=ExactTaskResolver(Source([record("t1"),record("t2")])).resolve(query(),actor_identity="caller")
assert result.state is ResolutionState.AMBIGUOUS
assert result.candidate_task_ids==("t1","t2")
def test_missing_selector_is_not_partial_match():
r=record("t1")
del r["plan"]["metadata"]["artifact_sha256"]
result=ExactTaskResolver(Source([r])).resolve(query(),actor_identity="caller")
assert result.state is ResolutionState.NOT_FOUND
def test_exact_task_id_still_revalidates_all_selectors():
q=query(task_id="t1")
result=ExactTaskResolver(Source([record("t1")])).resolve(q,actor_identity="caller")
assert result.state is ResolutionState.RESOLVED
bad=query(task_id="t1",candidate_oid="other")
result=ExactTaskResolver(Source([record("t1")])).resolve(bad,actor_identity="caller")
assert result.state is ResolutionState.NOT_FOUND
def test_weak_query_rejected():
q=ResolutionQuery(repository="aios/aios",base_oid="base")
result=ExactTaskResolver(Source([])).resolve(q,actor_identity="caller")
assert result.state is ResolutionState.INVALID_QUERY
@@ -0,0 +1,166 @@
from dataclasses import dataclass
import pytest
from app.task_plane.governance_admission import AdmissionOperation, GovernanceFacts
from app.task_plane.governance_checkpoint import JsonCheckpointStore, RetryBudget
from app.task_plane.governance_live_driver import GovernanceLiveDriver
from app.task_plane.governance_recovery_controller import (
GovernanceEvent,
GovernanceRecoveryController,
RecoveryReceipt,
)
from app.task_plane.governance_resolver import ResolutionQuery, ResolutionState
from app.task_plane.site_adapters import (
GovernedAuthoritySnapshot,
LiveBindingViolation,
LiveEventSubscriptionAdapter,
LiveGovernanceEvent,
LiveGovernanceFactsAdapter,
LiveSurfaceUnavailable,
LiveTaskResolverAdapter,
)
REPO="aios/aios"
BASE="5cf20128d8be2d10aef87b09f8f7a2aecb2000f3"
ART="7bd5e0c057b0ddf98a7dc81b0694599cbeda298179036788dec02a24922f4d74"
ACTOR="principal:recovery"
QUERY=ResolutionQuery(repository=REPO,base_oid=BASE,artifact_sha256=ART)
def facts(**kw):
body=dict(
governed_writes_enabled=True,
task_principal_writes_enabled=True,
recovery_execution_enabled=True,
recovery_shadow_mode=False,
authenticated_principal_id=ACTOR,
target_branch="agent/v41",
branch_protection_known=True,
expected_branch_oid=BASE,
generation=1,
task_lease_id="tl",
task_principal_id="tp",
writer_lease_id="wl",
accepted_file_allowlist=("a.py",),
acceptance_revision="rev1",
)
body.update(kw)
return GovernanceFacts(**body)
class ReadyTasks:
def get_task(self,task_id,*,actor_identity):
return self.list_tasks(actor_identity=actor_identity)[0] if task_id=="task-v41" else None
def list_tasks(self,*,actor_identity):
return [{
"task_id":"task-v41",
"mission_id":"m",
"root_task_id":"r",
"parent_task_id":"p",
"owner_principal_id":"owner",
"plan":{"metadata":{
"repository":REPO,
"base_oid":BASE,
"artifact_sha256":ART,
}}
}]
class AtomicFacts:
def __init__(self, value=None):
self.value=value or facts()
self.calls=[]
def read_snapshot(self,*,task_id,actor_identity,operation):
self.calls.append((task_id,actor_identity,operation))
return GovernedAuthoritySnapshot(
task_id=task_id,
actor_identity=actor_identity,
operation=operation,
snapshot_revision="snap-1",
facts=self.value,
)
def test_a1_rejects_controller_actor_mismatch():
adapter=LiveTaskResolverAdapter(
client=ReadyTasks(),actor_identity=ACTOR,bound_repository=REPO)
with pytest.raises(LiveBindingViolation,match="actor_identity"):
adapter.resolve(QUERY,actor_identity="principal:other")
def test_a1_reuses_exact_resolver_and_resolves_only_exact_candidate():
adapter=LiveTaskResolverAdapter(
client=ReadyTasks(),actor_identity=ACTOR,bound_repository=REPO)
out=adapter.resolve(QUERY,actor_identity=ACTOR)
assert out.state is ResolutionState.RESOLVED
assert out.resolved.task_id=="task-v41"
def test_a2_reads_one_candidate_specific_revision_consistent_snapshot():
client=AtomicFacts()
adapter=LiveGovernanceFactsAdapter(client=client,actor_identity=ACTOR)
out=adapter.snapshot(
task_id="task-v41",actor_identity=ACTOR,
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
assert out.writer_lease_id=="wl"
assert client.calls==[
("task-v41",ACTOR,AdmissionOperation.RECOVER_EXISTING_CHILD)
]
def test_a2_rejects_snapshot_binding_mismatch():
class Bad(AtomicFacts):
def read_snapshot(self,**kw):
snap=super().read_snapshot(**kw)
return GovernedAuthoritySnapshot(
task_id="other",
actor_identity=snap.actor_identity,
operation=snap.operation,
snapshot_revision=snap.snapshot_revision,
facts=snap.facts,
)
adapter=LiveGovernanceFactsAdapter(client=Bad(),actor_identity=ACTOR)
with pytest.raises(LiveBindingViolation,match="task mismatch"):
adapter.snapshot(
task_id="task-v41",actor_identity=ACTOR,
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
def test_a3_translates_live_event_to_core_event():
seen=[]
class Sub:
def subscribe(self,callback):
callback(LiveGovernanceEvent(
kind="LEASE_CHANGED",revision="42",occurred_at="now"))
def close(self): pass
adapter=LiveEventSubscriptionAdapter(lambda event: seen.append(event),Sub())
adapter.start()
assert len(seen)==1
assert isinstance(seen[0],GovernanceEvent)
assert (seen[0].kind,seen[0].revision)==("LEASE_CHANGED","42")
def test_live_driver_persists_surface_unavailable_without_escalation(tmp_path):
store=JsonCheckpointStore(tmp_path/"ck.json")
resolver=LiveTaskResolverAdapter()
controller=GovernanceRecoveryController(
goal_id="g",query=QUERY,actor_identity=ACTOR,
resolver=resolver,
facts_provider=LiveGovernanceFactsAdapter(),
executor=None,
checkpoint_store=store,
retry_budget=RetryBudget(unchanged_threshold=3),
)
driver=GovernanceLiveDriver(controller)
out=driver.on_event(GovernanceEvent("CHANGE","1","now"))
assert out.blocker_code=="LIVE_SURFACE_UNAVAILABLE"
assert out.next_legal_action=="WAIT_DEPENDENCY"
assert store.load().mechanism_repair_required is False
def test_live_driver_binding_violation_is_mechanism_repair(tmp_path):
store=JsonCheckpointStore(tmp_path/"ck.json")
resolver=LiveTaskResolverAdapter(
client=ReadyTasks(),actor_identity=ACTOR,bound_repository=REPO)
controller=GovernanceRecoveryController(
goal_id="g",query=QUERY,actor_identity="principal:wrong",
resolver=resolver,
facts_provider=LiveGovernanceFactsAdapter(),
executor=None,
checkpoint_store=store,
)
out=GovernanceLiveDriver(controller).on_event(
GovernanceEvent("CHANGE","1","now"))
assert out.blocker_code=="LIVE_BINDING_VIOLATION"
assert out.next_legal_action=="REPAIR_MECHANISM"
assert store.load().mechanism_repair_required is True
@@ -0,0 +1,77 @@
from app.task_plane.governance_admission import GovernanceFacts
from app.task_plane.governance_checkpoint import JsonCheckpointStore
from app.task_plane.governance_recovery_controller import (
GovernanceEvent,GovernanceRecoveryController,RecoveryReceipt
)
from app.task_plane.governance_resolver import ExactTaskResolver,ResolutionQuery
REPO="aios/aios"; BASE="base"; ART="artifact"; ACTOR="principal"
QUERY=ResolutionQuery(repository=REPO,base_oid=BASE,artifact_sha256=ART)
class Tasks:
def get_task(self,task_id,*,actor_identity): return self.list_tasks(actor_identity=actor_identity)[0]
def list_tasks(self,*,actor_identity):
return [{"task_id":"t","mission_id":"m","root_task_id":"r","parent_task_id":"p",
"plan":{"metadata":{"repository":REPO,"base_oid":BASE,"artifact_sha256":ART}}}]
def facts(gen=1,writer="wl"):
return GovernanceFacts(
True,True,True,False,ACTOR,"agent/x",True,BASE,gen,
"tl","tp",writer,("a.py",),"rev")
class MutableFacts:
def __init__(self,value): self.value=value
def snapshot(self,**kw): return self.value
class ReconcileExecutor:
def __init__(self):
self.recover_calls=[]; self.reconcile_calls=[]; self.fail_first=True
def recover_existing_child(self,**kw):
self.recover_calls.append(kw)
if self.fail_first:
self.fail_first=False
raise ConnectionError("lost")
return RecoveryReceipt("unexpected",kw["task_id"],True)
def reconcile_recovery_result(self,**kw):
self.reconcile_calls.append(kw)
return RecoveryReceipt("receipt-from-reconcile",kw["task_id"],True)
class NoReconcileExecutor:
def __init__(self): self.recover_calls=[]
def recover_existing_child(self,**kw):
self.recover_calls.append(kw)
raise ConnectionError("lost")
def event(rev): return GovernanceEvent("CHANGE",rev,"now")
def test_unknown_result_reconciles_prior_key_before_reexecution_even_if_facts_drift(tmp_path):
fp=MutableFacts(facts())
ex=ReconcileExecutor()
ctl=GovernanceRecoveryController(
goal_id="g",query=QUERY,actor_identity=ACTOR,
resolver=ExactTaskResolver(Tasks()),facts_provider=fp,executor=ex,
checkpoint_store=JsonCheckpointStore(tmp_path/"ck.json"))
first=ctl.on_event(event("1"))
assert first.status=="RESULT_UNKNOWN"
prior=ctl.checkpoints.load().recovery_idempotency_key
fp.value=facts(gen=2,writer="wl2")
second=ctl.on_event(event("2"))
assert second.status=="RECOVERED"
assert len(ex.recover_calls)==1
assert ex.reconcile_calls[0]["idempotency_key"]==prior
def test_unknown_result_with_context_drift_and_no_reconcile_does_not_second_mutate(tmp_path):
fp=MutableFacts(facts())
ex=NoReconcileExecutor()
ctl=GovernanceRecoveryController(
goal_id="g",query=QUERY,actor_identity=ACTOR,
resolver=ExactTaskResolver(Tasks()),facts_provider=fp,executor=ex,
checkpoint_store=JsonCheckpointStore(tmp_path/"ck.json"))
first=ctl.on_event(event("1"))
assert first.status=="RESULT_UNKNOWN"
key=ctl.checkpoints.load().recovery_idempotency_key
fp.value=facts(gen=9,writer="new")
second=ctl.on_event(event("2"))
assert second.status=="RESULT_UNKNOWN"
assert len(ex.recover_calls)==1
assert ctl.checkpoints.load().recovery_idempotency_key==key
@@ -0,0 +1,415 @@
import tempfile
import unittest
from pathlib import Path
from app.task_plane.governance_control_kernel import (
ActionLedger, ConsistencyBarrier, DiscoveryObservation, ManifestStore, StaleManifest
)
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
BranchReadback,
ClaimReceipt,
EpochStatus,
GovernedWriteReadiness,
GovernedWriterAdapters,
GovernedWriterAutomation,
GovernedWriterRequest,
GovernedWriterUnavailable,
JobBindingReceipt,
OwnerActionRequired,
OwnerAuthorizationPolicy,
RecoveryAuthorization,
TaskLeaseReceipt,
TaskPrincipalReceipt,
TaskState,
WriterLeaseReceipt,
VerifiedOwnerAuthorization,
)
def obs(worker="W1"):
return DiscoveryObservation.build(
worker=worker,
runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1",
activation_digest="activation-A",
epoch_id="epoch-old",
policy_digest="policy-A",
plan_provider_root="/plans/A",
task_id="task-A",
grant_id="grant-A",
facts={},
observed_at=100.0,
)
def make_manifest_store(td):
store=ManifestStore(Path(td)/"manifest.db")
b=ConsistencyBarrier(["W1"]); b.submit(obs()); m=b.freeze(); mid=store.freeze(m)
return store,m,mid
def owner_policy(now=100.0):
return OwnerAuthorizationPolicy(
authorization_id="owner-policy-1",
subject="agent-1",
repository="aios/aios",
allowed_branch_prefixes=("agent/",),
allowed_capabilities=frozenset({
"code.read","code.write","test.execute",
"governed.write.file","governed.write.branch",
"task.principal.issue","writer.lease","git.read","git.test",
}),
risk_ceiling="medium",
trust_domain="aios.owner",
revocation_epoch=0,
not_before=0.0,
expires_at=None,
)
def writer_request(mid):
return GovernedWriterRequest(
subject="agent-1",
task_id="task-A",
repository="aios/aios",
target_branch="agent/governance-kernel",
risk="medium",
generation=1,
base_oid="base-oid",
required_capabilities=frozenset({
"code.read","code.write","governed.write.file",
"governed.write.branch","task.principal.issue","writer.lease",
}),
file_allowlist=("active-runtime/app/task_plane/governed_writer_automation.py",),
plan_digest="plan-1",
manifest_id=mid,
runtime_fingerprint="runtime-A",
process_start_identity="pid:100:start:1",
)
class FakeOwnerAuthority:
def __init__(self, policy=None):
self.policy=policy or owner_policy()
self.calls=0
def verify(self, *, authorization_ref, request):
self.calls+=1
return VerifiedOwnerAuthorization(
policy=self.policy,
authorization_ref=authorization_ref,
verification_receipt="owner-verify-receipt-1",
verified_at=100.0,
)
class FakeSigner:
def __init__(self):
self.calls=0
def reissue_epoch(self, *, owner_policy, request, prior_epoch, minimum_remaining_ttl, idempotency_key):
self.calls+=1
return EpochStatus(
epoch_id=f"epoch-{self.calls}",
scope_fingerprint=request.scope_fingerprint,
expires_at=100.0+minimum_remaining_ttl+3600,
policy_digest="policy-A",
)
class FakeExecution:
def __init__(self, status="QUEUED"):
self.task=TaskState("task-A",status,1,None)
self.recoveries=0; self.claims=0; self.heartbeats=0; self.leases=0
def read_task(self, task_id):
return self.task
def apply_recovery(self, *, task, authorization, idempotency_key):
self.recoveries+=1
self.task=TaskState(task.task_id,"QUEUED",task.revision+1,None)
return self.task
def claim(self, *, task, actor, idempotency_key):
if task.status!="QUEUED":
raise RuntimeError("not claimable")
self.claims+=1
self.task=TaskState(task.task_id,"CLAIMED",task.revision+1,actor)
return ClaimReceipt("claim-1",task.task_id,actor,7)
def heartbeat(self, *, claim):
self.heartbeats+=1
def issue_task_lease(self, *, claim, idempotency_key):
self.leases+=1
return TaskLeaseReceipt("tl-1",claim.task_id,"secret-ref-not-token",5000.0,claim.fencing)
class FakePrincipalIssuer:
def __init__(self, drop_capability=None):
self.calls=0; self.drop=drop_capability
def issue(self, *, lease, requested_capabilities, idempotency_key):
self.calls+=1
caps=set(requested_capabilities)
if self.drop:
caps.discard(self.drop)
return TaskPrincipalReceipt("principal-1",lease.task_id,frozenset(caps),5000.0)
class FakeJobs:
def __init__(self):
self.calls=0
def read_or_reconcile(self, *, task_id, principal, idempotency_key):
self.calls+=1
return JobBindingReceipt("job-1",task_id,"aios/aios","agent/governance-kernel",1)
class FakeGit:
def __init__(self, exists=False, head=None):
self.exists=exists; self.head=head; self.calls=0
def read_branch(self, *, repository, target_branch):
self.calls+=1
return BranchReadback(self.exists,self.head)
class FakeWriterLease:
def __init__(self):
self.calls=0; self.last=None
def acquire(self, **kwargs):
self.calls+=1; self.last=kwargs
return WriterLeaseReceipt(
writer_lease_id="wl-1",
task_id=kwargs["task_id"],
job_id=kwargs["job"].job_id,
repository=kwargs["repository"],
target_branch=kwargs["target_branch"],
branch_mode=kwargs["branch_mode"],
expected_branch_oid=kwargs["expected_branch_oid"],
generation=kwargs["generation"],
expires_at=5000.0,
fencing=9,
)
def adapters(execution=None, git=None, principal=None, owner_authority=None):
return GovernedWriterAdapters(
signer=FakeSigner(),
execution=execution or FakeExecution(),
principal_issuer=principal or FakePrincipalIssuer(),
jobs=FakeJobs(),
git=git or FakeGit(),
writer_lease=FakeWriterLease(),
owner_authority=owner_authority,
)
class ReadinessTests(unittest.TestCase):
def test_legacy_global_false_is_required_invariant_not_blocker(self):
r=GovernedWriteReadiness(
legacy_global_writes_enabled=False,
task_principal_writes_enabled=True,
task_principal_verifier_ready=True,
)
self.assertTrue(r.governed_write_path_open)
self.assertEqual(r.blockers,())
def test_ambiguous_writes_enabled_false_is_ignored(self):
r=GovernedWriteReadiness.from_runtime_status({
"writes_enabled":False,
"AIOS_GITOPS_WRITES_ENABLED":False,
"legacy_global_writes_enabled":False,
"task_principal_writes_enabled":True,
"task_principal_readiness":{"ready":True},
"governed_write_blockers":[],
})
self.assertTrue(r.governed_write_path_open)
def test_legacy_global_true_blocks_governed_path(self):
r=GovernedWriteReadiness(
legacy_global_writes_enabled=True,
task_principal_writes_enabled=True,
task_principal_verifier_ready=True,
)
self.assertFalse(r.governed_write_path_open)
self.assertIn("LEGACY_GLOBAL_WRITES_MUST_REMAIN_DISABLED",r.blockers)
class OwnerPolicyTests(unittest.TestCase):
def test_owner_policy_rejects_scope_expansion(self):
with tempfile.TemporaryDirectory() as td:
_,_,mid=make_manifest_store(td)
req=writer_request(mid)
bad=OwnerAuthorizationPolicy(
authorization_id="bad",subject="agent-1",repository="aios/aios",
allowed_branch_prefixes=("agent/",),
allowed_capabilities=frozenset({"code.read"}),
risk_ceiling="medium",trust_domain="aios.owner",
revocation_epoch=0,not_before=0.0
)
with self.assertRaises(AuthorizationDenied):
bad.validate(req,now=100.0)
class AutomationTests(unittest.TestCase):
def _setup(self,status="QUEUED",git=None,principal=None):
td=tempfile.TemporaryDirectory()
ms,_,mid=make_manifest_store(td.name)
al=ActionLedger(Path(td.name)/"actions.db")
ex=FakeExecution(status)
ad=adapters(execution=ex,git=git,principal=principal)
auto=GovernedWriterAutomation(manifest_store=ms,action_ledger=al)
req=writer_request(mid)
ready=GovernedWriteReadiness(False,True,True,())
return td,ms,al,ex,ad,auto,req,ready
def test_owner_authorized_agent_acquires_writer_create_new(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
session=auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(session.writer_lease_id,"wl-1")
self.assertEqual(session.branch_mode,"CREATE_NEW")
self.assertEqual(session.expected_branch_oid,"base-oid")
self.assertEqual(ex.claims,1); self.assertEqual(ex.heartbeats,1); self.assertEqual(ex.leases,1)
self.assertEqual(ad.signer.calls,1)
finally:
td.cleanup()
def test_existing_branch_uses_live_head_for_cas(self):
td,ms,al,ex,ad,auto,req,ready=self._setup(git=FakeGit(True,"live-head"))
try:
session=auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(session.branch_mode,"CONTINUE_EXISTING")
self.assertEqual(session.expected_branch_oid,"live-head")
self.assertEqual(ad.writer_lease.last["expected_branch_oid"],"live-head")
finally:
td.cleanup()
def test_valid_epoch_is_reused_without_signer_call(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
epoch=EpochStatus("e",req.scope_fingerprint,5000.0,"policy-A")
auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=epoch,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ad.signer.calls,0)
finally:
td.cleanup()
def test_expired_epoch_same_scope_is_auto_reissued(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
epoch=EpochStatus("old",req.scope_fingerprint,101.0,"policy-A")
session=auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=epoch,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ad.signer.calls,1)
self.assertEqual(session.epoch_id,"epoch-1")
finally:
td.cleanup()
def test_changed_epoch_scope_requires_owner(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
epoch=EpochStatus("old","sha256:different",5000.0,"policy-A")
with self.assertRaises(OwnerActionRequired):
auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=epoch,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
finally:
td.cleanup()
def test_expired_task_without_signed_recovery_requires_owner(self):
td,ms,al,ex,ad,auto,req,ready=self._setup(status="EXPIRED")
try:
with self.assertRaises(OwnerActionRequired):
auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ex.claims,0)
finally:
td.cleanup()
def test_expired_task_with_preapproved_recovery_is_automatic(self):
td,ms,al,ex,ad,auto,req,ready=self._setup(status="EXPIRED")
try:
auth=RecoveryAuthorization("owner-signed-recovery-1","task-A","TERMINAL_RECOVERY",5000.0)
session=auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,recovery_authorization=auth,
now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ex.recoveries,1)
self.assertEqual(ex.claims,1)
self.assertEqual(session.writer_lease_id,"wl-1")
finally:
td.cleanup()
def test_stale_manifest_blocks_before_any_side_effect(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
bad=GovernedWriterRequest(**{
**req.__dict__,"runtime_fingerprint":"runtime-B"
})
with self.assertRaises(StaleManifest):
auto.acquire(
request=bad,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ad.signer.calls,0); self.assertEqual(ex.claims,0)
finally:
td.cleanup()
def test_principal_capability_shortfall_blocks_writer(self):
td,ms,al,ex,ad,auto,req,ready=self._setup(
principal=FakePrincipalIssuer(drop_capability="writer.lease")
)
try:
with self.assertRaises(GovernedWriterUnavailable):
auto.acquire(
request=req,owner_policy=owner_policy(),readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ad.writer_lease.calls,0)
finally:
td.cleanup()
def test_production_entry_requires_verified_owner_authorization(self):
td,ms,al,ex,ad,auto,req,ready=self._setup()
try:
with self.assertRaises(AuthorizationDenied):
auto.acquire_verified(
request=req,authorization_ref="owner-ref-1",readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
oa=FakeOwnerAuthority()
ad.owner_authority=oa
session=auto.acquire_verified(
request=req,authorization_ref="owner-ref-1",readiness=ready,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(oa.calls,1)
self.assertEqual(session.writer_lease_id,"wl-1")
finally:
td.cleanup()
def test_readiness_blocker_stops_before_signer(self):
td,ms,al,ex,ad,auto,req,_=self._setup()
try:
blocked=GovernedWriteReadiness(False,False,True,())
with self.assertRaises(GovernedWriterUnavailable):
auto.acquire(
request=req,owner_policy=owner_policy(),readiness=blocked,
epoch=None,adapters=ad,now=100.0,minimum_remaining_ttl=600,
)
self.assertEqual(ad.signer.calls,0)
finally:
td.cleanup()
if __name__=="__main__":
unittest.main()
@@ -0,0 +1,624 @@
"""生产入口(production entrypoint)集成测试:T2/T3 GovernedWriterAutomation。
与 V2 单元测试(test_governed_writer_automation.py,15 项)的区别:这里断言的是
**真实生产入口** `GovernedWriterAutomation.acquire_verified()` 的端到端行为,符号
全部从生产模块直接导入并真实执行——Owner 授权经正式 `OwnerAuthorizationAuthority`
验证后,Agent 在 `legacy_global_writes_enabled=false` 前提下自动走完
claim → TaskLease → TaskPrincipal → job → live branch CAS → WriterLease。
覆盖 V2 需求书强调的语义(不得因 `AIOS_GITOPS_WRITES_ENABLED=false` 宣布 writer
blocked):
- T2 生产入口:`acquire_verified()` 必须经正式 authority 返回 verification receipt;
未配置 authority → fail-closed;未经授权本地 policy 不构成 proof。
- T3 协议路径:legacy/global writes=false 不阻塞;ambiguous writes_enabled=false
被忽略;scope 扩张 fail-closed→Owner;同 scope 过期 epoch 经 signer JIT 重签;
QUEUED task 全自动达到 WriterLease;EXPIRED 无预签 recovery→Owner;
EXPIRED 有预签 recovery→自动恢复;live branch HEAD 决定 CAS;
能力不足阻断 WriterLease;stale manifest 在任何副作用前阻断。
Windows 说明:本文件不使用 `with tempfile.TemporaryDirectory()`,而是
`tempfile.mkdtemp()` + tearDown 中 `gc.collect()` 再 `rmtree(ignore_errors=True)`,
规避清理期 WinError 32。
"""
from __future__ import annotations
import gc
import shutil
import tempfile
import time
import unittest
from pathlib import Path
from app.task_plane.governance_control_kernel import (
ActionLedger,
ConsistencyBarrier,
ManifestStore,
DiscoveryObservation,
)
from app.task_plane.governed_writer_automation import (
AuthorizationDenied,
EpochStatus,
GovernedWriteReadiness,
GovernedWriterAdapters,
GovernedWriterAutomation,
GovernedWriterRequest,
GovernedWriterSession,
GovernedWriterState,
GovernedWriterUnavailable,
JobBindingReceipt,
OwnerActionRequired,
OwnerAuthorizationPolicy,
RecoveryAuthorization,
TaskLeaseReceipt,
TaskPrincipalReceipt,
TaskState,
VerifiedOwnerAuthorization,
WriterLeaseReceipt,
)
# ---------------------------------------------------------------------------
# 固定身份(与 V2 / S6 关联,仅测试用)
# ---------------------------------------------------------------------------
SUBJECT = "aios-autonomous-admin"
REPO = "aios/aios"
BRANCH = "agent/autonomy-v2-governed-writer-verify-20260917"
TASK = "task-gw-verify-0001"
BASE_OID = "5cf20128d8be2d10aef87b09f8f7a2aecb2000f3"
GENERATION = 1
RUNTIME = "runtime-gw-verify"
PROCESS_START = "pid:gw:start:1"
PLAN_DIGEST = "sha256:plan-gw-verify"
CAPS = frozenset(
{
"code.write",
"governed.write.branch",
"governed.write.file",
"test.execute",
"writer.lease",
"task.principal.issue",
}
)
FILES = (
"active-runtime/app/task_plane/governance_control_kernel.py",
"active-runtime/app/task_plane/governed_writer_automation.py",
)
def make_request(**overrides) -> GovernedWriterRequest:
body = dict(
subject=SUBJECT,
task_id=TASK,
repository=REPO,
target_branch=BRANCH,
risk="medium",
generation=GENERATION,
base_oid=BASE_OID,
required_capabilities=CAPS,
file_allowlist=FILES,
plan_digest=PLAN_DIGEST,
manifest_id="m",
runtime_fingerprint=RUNTIME,
process_start_identity=PROCESS_START,
)
body.update(overrides)
return GovernedWriterRequest(**body)
def make_policy(**overrides) -> OwnerAuthorizationPolicy:
body = dict(
authorization_id="azn-gw-verify",
subject=SUBJECT,
repository=REPO,
allowed_branch_prefixes=("agent/",),
allowed_capabilities=CAPS,
risk_ceiling="high",
trust_domain="aios.owner",
revocation_epoch=1,
not_before=0.0,
expires_at=None,
)
body.update(overrides)
return OwnerAuthorizationPolicy(**body)
def make_readiness(**overrides) -> GovernedWriteReadiness:
body = dict(
legacy_global_writes_enabled=False,
task_principal_writes_enabled=True,
task_principal_verifier_ready=True,
governed_write_blockers=(),
)
body.update(overrides)
return GovernedWriteReadiness(**body)
# ---------------------------------------------------------------------------
# 生产适配器的忠实实现(非 mock 内部,而是模拟真实协议端点的行为)
# ---------------------------------------------------------------------------
class _VerifyingAuthority:
"""OwnerAuthorizationAuthority 生产实现:解析 authorization_ref → 验证 → receipt。"""
def __init__(self, policy: OwnerAuthorizationPolicy, *, verify: bool = True):
self.policy = policy
self.verify_ok = verify
self.calls: list[dict] = []
def verify(self, *, authorization_ref: str, request: GovernedWriterRequest):
self.calls.append({"ref": authorization_ref, "task": request.task_id})
# 授权引用必须是已登记的正式引用;本地 policy 本身不构成 proof。
if authorization_ref != self.policy.authorization_id:
raise AuthorizationDenied("unknown authorization reference")
if not self.verify_ok:
raise AuthorizationDenied("authority declined verification")
return VerifiedOwnerAuthorization(
policy=self.policy,
authorization_ref=authorization_ref,
verification_receipt="receipt:gwa-verified:" + authorization_ref,
verified_at=time.time(),
)
class _Signer:
"""TrustedSignerAdapter 生产实现:JIT 重签同 scope epoch。"""
def __init__(self, *, fail: bool = False):
self.calls: list[dict] = []
self.fail = fail
self.counter = 0
def reissue_epoch(
self, *, owner_policy, request, prior_epoch, minimum_remaining_ttl, idempotency_key
):
self.calls.append(
{
"scope": request.scope_fingerprint,
"prior": None if prior_epoch is None else prior_epoch.epoch_id,
"key": idempotency_key,
}
)
if self.fail:
raise RuntimeError("signer unavailable")
self.counter += 1
return EpochStatus(
epoch_id=f"epoch-jit-{self.counter}",
scope_fingerprint=request.scope_fingerprint,
expires_at=time.time() + 7200,
policy_digest=owner_policy.policy_fingerprint,
)
class _ExecutionPlane:
"""ExecutionPlane 生产实现:QUEUED 自动入链,可模拟 EXPIRED + recovery。"""
def __init__(self, initial: TaskState):
self.task = initial
self.recovery_applied = False
self.claims: list = []
self.heartbeats: list = []
def read_task(self, task_id):
return self.task
def apply_recovery(self, *, task, authorization, idempotency_key):
self.recovery_applied = True
self.task = TaskState(task_id=task.task_id, status="QUEUED", revision=task.revision + 1)
return self.task
def claim(self, *, task, actor, idempotency_key):
self.claims.append({"actor": actor, "key": idempotency_key})
# 实现 ClaimReceipt 的 shape,但这里按生产真实契约返回带 fencing 的凭证
from types import SimpleNamespace
return SimpleNamespace(claim_id="claim-1", task_id=task.task_id, actor=actor, fencing=1)
def heartbeat(self, *, claim):
self.heartbeats.append(claim)
def issue_task_lease(self, *, claim, idempotency_key):
return TaskLeaseReceipt(
task_lease_id="tl-1",
task_id=claim.task_id,
token_ref="token-ref-1",
expires_at=time.time() + 3600,
fencing=claim.fencing,
)
class _PrincipalIssuer:
def __init__(self, *, shortfall: bool = False):
self.calls: list = []
self.shortfall = shortfall
def issue(self, *, lease, requested_capabilities, idempotency_key):
self.calls.append({"lease": lease.task_lease_id, "caps": sorted(requested_capabilities)})
granted = requested_capabilities
if self.shortfall:
granted = frozenset(c for c in requested_capabilities if not c.startswith("governed"))
return TaskPrincipalReceipt(
principal_id="tp-1",
task_id=lease.task_id,
capabilities=frozenset(granted),
expires_at=time.time() + 3600,
)
class _Jobs:
def __init__(self, *, mismatch: bool = False):
self.calls: list = []
self.mismatch = mismatch
def read_or_reconcile(self, *, task_id, principal, idempotency_key):
self.calls.append({"task": task_id, "principal": principal.principal_id})
return JobBindingReceipt(
job_id="job-1",
task_id=task_id,
repository=("other/repo" if self.mismatch else REPO),
target_branch=(BRANCH if not self.mismatch else "agent/other"),
generation=GENERATION,
)
class _Git:
def __init__(self, *, exists: bool = False, head: str | None = None):
self.exists = exists
self.head = head
def read_branch(self, *, repository, target_branch):
from types import SimpleNamespace
return SimpleNamespace(
exists=self.exists and self.head is not None,
head_oid=self.head,
)
class _WriterLease:
def __init__(self, *, cas_mismatch: bool = False):
self.calls: list = []
self.cas_mismatch = cas_mismatch
def acquire(
self, *, task_id, job, principal, repository, target_branch, branch_mode,
base_oid, expected_branch_oid, generation, idempotency_key,
):
self.calls.append(
{"branch_mode": branch_mode, "expected": expected_branch_oid, "base": base_oid}
)
eff = ("other" if self.cas_mismatch else expected_branch_oid)
return WriterLeaseReceipt(
writer_lease_id="wl-1",
task_id=task_id,
job_id=job.job_id,
repository=repository,
target_branch=target_branch,
branch_mode=branch_mode,
expected_branch_oid=eff,
generation=generation,
expires_at=time.time() + 3600,
fencing=1,
)
class _TempDirTestCase(unittest.TestCase):
def setUp(self):
self.tmp = Path(tempfile.mkdtemp(prefix="aios-gov-gw-entry-"))
def tearDown(self):
gc.collect()
shutil.rmtree(self.tmp, ignore_errors=True)
def _automation(self):
store = ManifestStore(self.tmp / "manifest.db")
ledger = ActionLedger(self.tmp / "actions.db")
return GovernedWriterAutomation(
manifest_store=store, action_ledger=ledger,
)
def _frozen_manifest_id(self, store: ManifestStore) -> str:
barrier = ConsistencyBarrier(["W1"])
barrier.submit(_gw_observation())
return store.freeze(barrier.freeze())
def _gw_observation(worker: str = "W1", **overrides) -> DiscoveryObservation:
data = dict(
worker=worker,
runtime_fingerprint=RUNTIME,
process_start_identity=PROCESS_START,
activation_digest="activation-gw",
epoch_id="epoch-gw",
policy_digest="policy-gw",
plan_provider_root="/plans/gw",
task_id=TASK,
grant_id="grant-gw",
facts={},
evidence_refs=(),
observed_at=100.0,
)
data.update(overrides)
return DiscoveryObservation.build(**data)
# ---------------------------------------------------------------------------
# T2:生产入口 acquire_verified() —— 必须经正式 Owner 授权 authority 验证
# ---------------------------------------------------------------------------
class T2OwnerAuthorizationEntryTests(_TempDirTestCase):
def test_t2_production_entry_requires_verified_owner_authorization(self):
"""未配置 OwnerAuthorizationAuthority → fail-closed,本地 policy 不构成 proof。"""
repo = make_policy()
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
adapters_no_authority = GovernedWriterAdapters(
signer=_Signer(), execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=None,
)
with self.assertRaises(AuthorizationDenied):
automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters_no_authority,
now=time.time(),
)
# 即使本地 policy 有效,也必须经 authority 验证并返回 receipt,否则拒绝
adapters_ref_missing = GovernedWriterAdapters(
signer=_Signer(), execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=_VerifyingAuthority(repo),
)
with self.assertRaises(AuthorizationDenied):
automation.acquire_verified(
request=request, authorization_ref="unknown-ref",
readiness=make_readiness(), epoch=None, adapters=adapters_ref_missing,
now=time.time(),
)
def test_t2_verified_owner_authorization_drives_full_writer_chain(self):
"""Owner 授权一次验证通过后,Agent 自动走完 claim→…→WriterLease。"""
repo = make_policy()
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
signer = _Signer()
authority = _VerifyingAuthority(repo)
branch = _Git() # branch 不存在 → CREATE_NEW,expected = base_oid
adapters = GovernedWriterAdapters(
signer=signer, execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=branch,
writer_lease=_WriterLease(), owner_authority=authority,
)
session: GovernedWriterSession = automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(), minimum_remaining_ttl=1800.0,
)
self.assertEqual(session.state, GovernedWriterState.WRITER_LEASED)
self.assertEqual(session.branch_mode, "CREATE_NEW")
self.assertEqual(session.expected_branch_oid, BASE_OID)
self.assertEqual(session.authorization_id, "azn-gw-verify")
self.assertEqual(session.task_id, TASK)
self.assertTrue(session.epoch_id.startswith("epoch-jit-"))
self.assertEqual(len(authority.calls), 1)
# Epoch 由 trusted signer JIT 重签(scope 未变,Agent 自主)
self.assertEqual(len(signer.calls), 1)
self.assertEqual(signer.calls[0]["scope"], request.scope_fingerprint)
# ---------------------------------------------------------------------------
# T3:协议路径语义 —— legacy/global writes 与 governed writer 解耦
# ---------------------------------------------------------------------------
class T3GovernedWriteReadinessSemanticsTests(_TempDirTestCase):
def test_t3_legacy_global_false_is_invariant_not_blocker(self):
"""legacy_global_writes_enabled=false 不阻断 governed writer,反而是不变式。"""
readiness = make_readiness()
self.assertTrue(readiness.governed_write_path_open)
self.assertEqual(readiness.blockers, ())
def test_t3_ambiguous_writes_enabled_is_ignored(self):
"""from_runtime_status 忽略 ambiguous writes_enabled / AIOS_GITOPS_WRITES_ENABLED。"""
r = GovernedWriteReadiness.from_runtime_status(
{
"AIOS_GITOPS_WRITES_ENABLED": False,
"writes_enabled": False,
"legacy_global_writes_enabled": False,
"task_principal_writes_enabled": True,
"task_principal_readiness": {"ready": True},
"blockers": [],
}
)
self.assertTrue(r.governed_write_path_open)
self.assertFalse(r.legacy_global_writes_enabled)
self.assertEqual(r.blockers, ())
def test_t3_legacy_global_true_blocks(self):
r = make_readiness(legacy_global_writes_enabled=True)
self.assertFalse(r.governed_write_path_open)
self.assertIn("LEGACY_GLOBAL_WRITES_MUST_REMAIN_DISABLED", r.blockers)
class T3EpochAndTaskDecisionTests(_TempDirTestCase):
def _adapters(self, *, task: TaskState | None = None, with_recovery=None):
authority = _VerifyingAuthority(make_policy())
return authority, GovernedWriterAdapters(
signer=_Signer(),
execution=_ExecutionPlane(task or TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=authority,
)
def test_t3_changed_epoch_scope_requires_owner(self):
"""同 scope 过期 epoch 才可 JIT;scope 改变不可静默继承旧授权。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
authority, adapters = self._adapters()
stale_scope = EpochStatus(
epoch_id="epoch-old",
scope_fingerprint="sha256:old-different-scope",
expires_at=time.time() + 7200,
policy_digest="p",
)
with self.assertRaises(OwnerActionRequired):
automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=stale_scope, adapters=adapters,
now=time.time(),
)
def test_t3_expired_epoch_same_scope_is_auto_reissued(self):
"""同 scope epoch 过期 → 经 trusted signer 自动 JIT 重签(非 Owner 边界)。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
signer = _Signer()
authority = _VerifyingAuthority(make_policy())
adapters = GovernedWriterAdapters(
signer=signer, execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=authority,
)
expired = EpochStatus(
epoch_id="epoch-expired",
scope_fingerprint=request.scope_fingerprint,
expires_at=time.time() - 100,
policy_digest="p",
)
session = automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=expired, adapters=adapters,
now=time.time(), minimum_remaining_ttl=1800.0,
)
self.assertEqual(session.state, GovernedWriterState.WRITER_LEASED)
self.assertNotEqual(session.epoch_id, "epoch-expired")
self.assertEqual(len(signer.calls), 1)
def test_t3_expired_task_without_signed_recovery_requires_owner(self):
"""EXPIRED terminal task 无预签 recovery → OWNER_BOUNDARY(区别于 epoch)。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
authority, adapters = self._adapters(
task=TaskState(TASK, "EXPIRED", 2),
)
with self.assertRaises(OwnerActionRequired):
automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(), recovery_authorization=None,
)
def test_t3_expired_task_with_preapproved_recovery_is_automatic(self):
"""EXPIRED + 预签 RecoveryAuthorization → 自动恢复为 QUEUED 并继续。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
authority = _VerifyingAuthority(make_policy())
plane = _ExecutionPlane(TaskState(TASK, "EXPIRED", 2))
adapters = GovernedWriterAdapters(
signer=_Signer(), execution=plane,
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=authority,
)
recovery = RecoveryAuthorization(
authorization_ref="recovery-azn-1",
task_id=TASK,
action="TERMINAL_RECOVERY",
expires_at=time.time() + 3600,
)
session = automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(), recovery_authorization=recovery,
)
self.assertEqual(session.state, GovernedWriterState.WRITER_LEASED)
self.assertTrue(plane.recovery_applied)
class T3BranchCasAndGuardTests(_TempDirTestCase):
def test_t3_existing_branch_uses_live_head_for_cas(self):
"""branch 已存在 → CONTINUE_EXISTING,live HEAD 决定 expected_branch_oid。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
live_head = "abcd1234livehead"
authority = _VerifyingAuthority(make_policy())
writer = _WriterLease()
adapters = GovernedWriterAdapters(
signer=_Signer(), execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(),
git=_Git(exists=True, head=live_head), writer_lease=writer,
owner_authority=authority,
)
session = automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(),
)
self.assertEqual(session.branch_mode, "CONTINUE_EXISTING")
self.assertEqual(session.expected_branch_oid, live_head)
self.assertEqual(writer.calls[0]["expected"], live_head)
def test_t3_principal_capability_shortfall_blocks_writer(self):
"""TaskPrincipal 能力不足 → 阻断 WriterLease。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
request = make_request(manifest_id=manifest_id)
authority = _VerifyingAuthority(make_policy())
adapters = GovernedWriterAdapters(
signer=_Signer(), execution=_ExecutionPlane(TaskState(TASK, "QUEUED", 0)),
principal_issuer=_PrincipalIssuer(shortfall=True), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=authority,
)
with self.assertRaises(GovernedWriterUnavailable):
automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(),
)
def test_t3_stale_manifest_blocks_before_any_side_effect(self):
"""stale runtime manifest → 在任何副作用前阻断。"""
automation = self._automation()
store = automation.manifest_store
manifest_id = self._frozen_manifest_id(store)
# 请求用错误 runtime_fingerprint → assert_current 抛 StaleManifest
request = make_request(manifest_id=manifest_id, runtime_fingerprint="stale-runtime")
from app.task_plane.governance_control_kernel import StaleManifest
authority = _VerifyingAuthority(make_policy())
signer = _Signer()
plane = _ExecutionPlane(TaskState(TASK, "QUEUED", 0))
adapters = GovernedWriterAdapters(
signer=signer, execution=plane,
principal_issuer=_PrincipalIssuer(), jobs=_Jobs(), git=_Git(),
writer_lease=_WriterLease(), owner_authority=authority,
)
with self.assertRaises(StaleManifest):
automation.acquire_verified(
request=request, authorization_ref="azn-gw-verify",
readiness=make_readiness(), epoch=None, adapters=adapters,
now=time.time(),
)
# 既无 claim、无心跳、无 epoch 副作用
self.assertEqual(signer.calls, [])
self.assertEqual(plane.claims, [])
self.assertEqual(plane.heartbeats, [])
if __name__ == "__main__":
unittest.main()
+9 -5
View File
@@ -13,15 +13,19 @@ import pytest
# Exercise the overlay exactly as deployment will see it: changed modules from
# this artifact plus unchanged dependencies from the checked-out repository.
_WORK = Path(__file__).resolve().parents[4]
_OVERLAY_APP = Path(__file__).resolve().parents[2] / "source" / "active-runtime" / "app"
_REPO_APP = _WORK / "repo" / "active-runtime" / "app"
_APP_DIR = Path(__file__).resolve().parents[2] / "active-runtime" / "app"
# Production resolves ``app`` via the checkout-root symlink ``app -> active-runtime/app``
# (and tests/conftest.py places active-runtime/app on sys.path). Keep the overlay
# intent (changed modules + unchanged dependencies from the checked-out repo) by
# pointing the ``app``/``app.task_plane`` namespace at the REAL active-runtime/app,
# never at non-existent source//repo/ dirs (which poisoned sys.modules and broke
# collection of the other app.*-qualified tests in the full suite).
_app = types.ModuleType("app")
_app.__path__ = [str(_OVERLAY_APP), str(_REPO_APP)]
_app.__path__ = [str(_APP_DIR)]
_app.__package__ = "app"
sys.modules["app"] = _app
_task_plane = types.ModuleType("app.task_plane")
_task_plane.__path__ = [str(_OVERLAY_APP / "task_plane"), str(_REPO_APP / "task_plane")]
_task_plane.__path__ = [str(_APP_DIR / "task_plane")]
_task_plane.__package__ = "app.task_plane"
sys.modules["app.task_plane"] = _task_plane
@@ -21,7 +21,7 @@ TASK="legacy-task"
def _install_stubs():
app=types.ModuleType("app"); app.__path__=[]; sys.modules["app"]=app
app=types.ModuleType("app"); app.__path__=[str(ROOT / "active-runtime" / "app")]; sys.modules["app"]=app
identity=types.ModuleType("app.aios_gitops_job_identity")
identity.legacy_task_storage_key=lambda value:value
identity.task_storage_key=lambda value:value
@@ -34,7 +34,7 @@ def _install_stubs():
auth.SignerRegistry=type("SignerRegistry",(),{})
sys.modules[auth.__name__]=auth
task_plane=types.ModuleType("app.task_plane"); task_plane.__path__=[]
task_plane=types.ModuleType("app.task_plane"); task_plane.__path__=[str(ROOT / "active-runtime" / "app" / "task_plane")]
sys.modules["app.task_plane"]=task_plane
la=types.ModuleType("app.task_plane.legacy_attachment")
la.ATTACH_EVIDENCE_SCHEMA="evidence"
@@ -46,14 +46,36 @@ def _install_stubs():
sys.modules[la.__name__]=la
# These stubs are only needed at import time of the runtime module. Pytest runs
# ``importlib`` mode in a single interpreter, so we must not permanently replace
# the real ``app.*`` namespace in ``sys.modules`` -- doing so breaks every later
# ``app.*`` import in the same suite (e.g. app.task_plane.schedulability). Snapshot
# and restore the affected entries around the module load.
_STUB_NAMES = [
"app",
"app.aios_gitops_job_identity",
"app.aios_single_host_authority",
"app.task_plane",
"app.task_plane.legacy_attachment",
]
def _load():
_install_stubs()
name="app.aios_legacy_attachment_runtime"
spec=importlib.util.spec_from_file_location(name,MODULE_PATH)
mod=importlib.util.module_from_spec(spec)
sys.modules[name]=mod
spec.loader.exec_module(mod)
return mod
saved={name: sys.modules.get(name) for name in _STUB_NAMES}
try:
_install_stubs()
name="app.aios_legacy_attachment_runtime"
spec=importlib.util.spec_from_file_location(name,MODULE_PATH)
mod=importlib.util.module_from_spec(spec)
sys.modules[name]=mod
spec.loader.exec_module(mod)
return mod
finally:
for n in _STUB_NAMES:
if saved[n] is None:
sys.modules.pop(n, None)
else:
sys.modules[n]=saved[n]
M=_load()
@@ -0,0 +1,354 @@
import hashlib
import tempfile
import time
import unittest
from pathlib import Path
from app.task_plane.governed_writer_automation import (
GovernedWriterRequest,
OwnerAuthorizationPolicy,
)
from app.task_plane.live_governed_writer_adapters import (
HttpResult,
LiveAgentTaskClient,
LiveBridgeClient,
LiveGovernedWriterExecutor,
LiveProtocolConfig,
LiveProtocolConflict,
LiveProtocolError,
LiveTaskPrincipalClient,
SecretHandleStore,
)
class RecordingTransport:
def __init__(self):
self.calls=[]
self.task_reads=0
def request(self,method,url,*,body=None,headers=None,timeout=20.0):
self.calls.append({
"method":method,"url":url,"body":body or {},
"headers":headers or {},"timeout":timeout,
})
path=url.split("127.0.0.1")[-1]
if method=="GET" and path.endswith(":18892/v2/runtime"):
return HttpResult(200,{
"legacy_global_writes_enabled":False,
"task_principal_writes_enabled":True,
"task_principal_readiness":{"ready":True},
"governed_write_blockers":[],
})
if method=="POST" and path.endswith(":18940/v1/agents/register"):
return HttpResult(201,{"agent_token":"agent-token-secret-12345"})
if method=="GET" and ":18940/v1/tasks/" in path:
self.task_reads+=1
status="QUEUED" if self.task_reads==1 else "RUNNING"
return HttpResult(200,{
"task_id":"task-live-1",
"status":status,
"revision":4+self.task_reads,
"deadline":time.time()+1200,
"lease_expires_at":time.time()+900,
})
if method=="POST" and path.endswith("/claim"):
return HttpResult(200,{
"status":"CLAIMED",
"claim_id":"claim-live-1",
"lease_id":"lease-live-1",
"lease_token":"lease-token-secret-12345",
"fencing_generation":9,
"receipt_revision":7,
"lease_expires_at":time.time()+900,
})
if method=="POST" and path.endswith("/heartbeat"):
return HttpResult(200,{"status":"RUNNING"})
if method=="POST" and path.endswith(":18946/v1/task-principal/assertions"):
operation=(body or {}).get("operation")
return HttpResult(201,{"assertion":f"assertion-for-{operation}"})
if method=="POST" and path.endswith(":18892/v2/writer-lease"):
return HttpResult(201,{
"schema":"aios.task-writer-lease.v2",
"task_id":"task-live-1",
"lease_id":"writer-lease-live-1",
"expected_branch_oid":(body or {}).get("expected_branch_oid"),
"expires_at":time.time()+900,
})
if method=="POST" and path.endswith(":18892/v2/commit-files"):
return HttpResult(201,{
"status":"COMMITTED",
"task_id":"task-live-1",
"commit_oid":"new-head-oid",
})
raise AssertionError(f"unexpected call: {method} {url} {body}")
class FakeGit:
def __init__(self,head=None):
self.head=head
self.sha={}
self.reads=0
def branch_head(self,branch):
self.reads+=1
return self.head
def file_sha256(self,oid,path):
return self.sha[path]
def policy():
return OwnerAuthorizationPolicy(
authorization_id="azn-owner-gw-v1",
subject="aios-autonomous-admin",
repository="aios/aios",
allowed_branch_prefixes=("agent/autonomy-",),
allowed_capabilities=frozenset({
"code.write","governed.write.branch","governed.write.file",
"test.execute","writer.lease","task.principal.issue",
}),
risk_ceiling="medium",
trust_domain="aios.owner",
revocation_epoch=1,
not_before=0.0,
expires_at=None,
revoked=False,
)
def request():
return GovernedWriterRequest(
subject="aios-autonomous-admin",
task_id="task-live-1",
repository="aios/aios",
target_branch="agent/autonomy-live-protocol",
risk="medium",
generation=1,
base_oid="5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
required_capabilities=frozenset({
"code.write","governed.write.branch","governed.write.file",
"test.execute","writer.lease","task.principal.issue",
}),
file_allowlist=("active-runtime/app/task_plane/x.py",),
plan_digest="sha256:plan-live",
manifest_id="sha256:manifest-live",
runtime_fingerprint="runtime-live",
process_start_identity="pid:live:start:1",
)
class LiveAgentTaskContractTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory()
self.control=Path(self.tmp.name)/"control.token"
self.control.write_text("control-token-secret-12345",encoding="utf-8")
self.transport=RecordingTransport()
self.secrets=SecretHandleStore()
self.cfg=LiveProtocolConfig(control_token_file=str(self.control))
def tearDown(self):
self.tmp.cleanup()
def test_claim_and_heartbeat_use_live_contract(self):
client=LiveAgentTaskClient(
self.cfg,transport=self.transport,secrets_store=self.secrets
)
task=client.read_task("task-live-1")
self.assertEqual(task.status,"QUEUED")
agent_id,agent_handle=client.register_agent(
task_id="task-live-1",
capabilities=["governed.write.branch"],
route="test",
)
claim=client.claim(
task_id="task-live-1",agent_token_handle=agent_handle
)
self.assertEqual(claim.claim_ref,"claim-live-1")
self.assertEqual(claim.fencing_generation,9)
running=client.heartbeat(
claim=claim,agent_token_handle=agent_handle,phase="test"
)
self.assertEqual(running.status,"RUNNING")
claim_call=[c for c in self.transport.calls if c["url"].endswith("/claim")][0]
self.assertEqual(claim_call["body"],{"lease_seconds":900})
self.assertTrue(claim_call["headers"]["Authorization"].startswith("Bearer "))
hb=[c for c in self.transport.calls if c["url"].endswith("/heartbeat")][0]
self.assertEqual(hb["body"]["lease_token"],"lease-token-secret-12345")
def test_principal_assertion_is_operation_scoped(self):
self.secrets.put("dummy","ignore")
handle=self.secrets.put("task-lease","lease-token-secret-12345")
principal=LiveTaskPrincipalClient(
self.cfg,transport=self.transport,secrets_store=self.secrets
)
assertion=principal.issue_assertion(
task_id="task-live-1",
lease_handle=handle,
operation="acquireWriterLease",
bridge_request={"schema":"aios.task-writer-lease.v2"},
request_id="req-1",
)
self.assertEqual(assertion,"assertion-for-acquireWriterLease")
call=self.transport.calls[-1]
self.assertEqual(call["body"]["schema"],"aios.task-principal-issue.v1")
self.assertEqual(call["body"]["operation"],"acquireWriterLease")
self.assertEqual(
call["headers"]["Authorization"],
"TaskLease lease-token-secret-12345",
)
class LiveBridgeContinuationTests(unittest.TestCase):
def setUp(self):
self.transport=RecordingTransport()
self.secrets=SecretHandleStore()
self.cfg=LiveProtocolConfig()
self.lease_handle=self.secrets.put(
"task-lease","lease-token-secret-12345"
)
self.principal=LiveTaskPrincipalClient(
self.cfg,transport=self.transport,secrets_store=self.secrets
)
self.bridge=LiveBridgeClient(
self.cfg,transport=self.transport,principal_client=self.principal
)
def test_writerlease_and_commit_use_same_continuation_key(self):
wl=self.bridge.acquire_writer_lease(
task_id="task-live-1",
job_id="job-live-1",
lease_handle=self.lease_handle,
repository="aios/aios",
target_branch="agent/autonomy-live-protocol",
generation=1,
base_oid="5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
expected_branch_oid="5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
branch_mode="CREATE_NEW",
continuation_key="gw-cont-fixed",
)
from app.task_plane.live_governed_writer_adapters import LiveWriterSession
session=LiveWriterSession(
authorization_id="azn-owner-gw-v1",
task_id="task-live-1",
agent_id="agent-live",
claim_ref="claim-live-1",
job_id="job-live-1",
branch_mode="CREATE_NEW",
expected_branch_oid=wl.expected_branch_oid,
writer_lease_id=wl.writer_lease_id,
continuation_key=wl.continuation_key,
principal_issue_schema=self.cfg.issue_schema,
lease_handle=self.lease_handle,
)
self.bridge.commit_files(
session=session,
repository="aios/aios",
target_branch="agent/autonomy-live-protocol",
generation=1,
files={"x.py":b"x=1\n"},
message="test",
)
wl_call=[c for c in self.transport.calls if c["url"].endswith("/v2/writer-lease")][0]
cm_call=[c for c in self.transport.calls if c["url"].endswith("/v2/commit-files")][0]
self.assertEqual(wl_call["body"]["idempotency_key"],"gw-cont-fixed")
self.assertEqual(cm_call["body"]["idempotency_key"],"gw-cont-fixed")
issue_ops=[
c["body"]["operation"] for c in self.transport.calls
if c["url"].endswith("/v1/task-principal/assertions")
]
self.assertEqual(issue_ops,["acquireWriterLease","commitFilesGoverned"])
class LiveExecutorTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory()
self.control=Path(self.tmp.name)/"control.token"
self.control.write_text("control-token-secret-12345",encoding="utf-8")
self.transport=RecordingTransport()
self.cfg=LiveProtocolConfig(control_token_file=str(self.control))
def tearDown(self):
self.tmp.cleanup()
def test_full_post_admission_chain_create_new(self):
ex=LiveGovernedWriterExecutor(
config=self.cfg,
owner_policy=policy(),
authorization_ref="azn-owner-gw-v1",
transport=self.transport,
)
ex.git=FakeGit(head=None)
session=ex.acquire(request=request(),job_id="job-live-1")
self.assertEqual(session.branch_mode,"CREATE_NEW")
self.assertEqual(
session.expected_branch_oid,
"5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
)
self.assertEqual(session.writer_lease_id,"writer-lease-live-1")
def test_nonqueued_task_fails_before_agent_registration(self):
class ExpiredTransport(RecordingTransport):
def request(self,method,url,*,body=None,headers=None,timeout=20.0):
if method=="GET" and ":18940/v1/tasks/" in url:
self.calls.append({"method":method,"url":url,"body":body or {},"headers":headers or {}})
return HttpResult(200,{
"task_id":"task-live-1","status":"EXPIRED",
"revision":2,"deadline":time.time()-10,
})
return super().request(method,url,body=body,headers=headers,timeout=timeout)
tr=ExpiredTransport()
ex=LiveGovernedWriterExecutor(
config=self.cfg,owner_policy=policy(),
authorization_ref="azn-owner-gw-v1",transport=tr,
)
ex.git=FakeGit()
with self.assertRaises(LiveProtocolConflict):
ex.acquire(request=request(),job_id="job-live-1")
self.assertFalse(any(c["url"].endswith("/v1/agents/register") for c in tr.calls))
def test_commit_byte_gate_and_readback(self):
ex=LiveGovernedWriterExecutor(
config=self.cfg,owner_policy=policy(),
authorization_ref="azn-owner-gw-v1",transport=self.transport,
)
fg=FakeGit(head=None)
ex.git=fg
session=ex.acquire(request=request(),job_id="job-live-1")
data=b"x=1\n"
sha=hashlib.sha256(data).hexdigest()
fg.head="0123456789abcdef0123456789abcdef01234567"
fg.sha={"active-runtime/app/task_plane/x.py":sha}
result=ex.commit_and_readback(
session=session,
request=request(),
files={"active-runtime/app/task_plane/x.py":data},
expected_sha256={"active-runtime/app/task_plane/x.py":sha},
message="test commit",
)
self.assertEqual(
result["branch_head"],
"0123456789abcdef0123456789abcdef01234567",
)
def test_commit_byte_gate_rejects_mismatch_before_bridge(self):
ex=LiveGovernedWriterExecutor(
config=self.cfg,owner_policy=policy(),
authorization_ref="azn-owner-gw-v1",transport=self.transport,
)
ex.git=FakeGit(head=None)
session=ex.acquire(request=request(),job_id="job-live-1")
count_before=len(self.transport.calls)
with self.assertRaises(LiveProtocolConflict):
ex.commit_and_readback(
session=session,
request=request(),
files={"active-runtime/app/task_plane/x.py":b"bad"},
expected_sha256={
"active-runtime/app/task_plane/x.py":"0"*64
},
message="test",
)
commit_calls=[
c for c in self.transport.calls[count_before:]
if c["url"].endswith("/v2/commit-files")
]
self.assertEqual(commit_calls,[])
if __name__=="__main__":
unittest.main()
@@ -147,7 +147,7 @@ def _fixture(tmp_path):
"goal_slug": "terminal-proof-test",
"original_acceptance": {
"schema": "acceptance.v1",
"conditions": [{"key": "C01"}, {"key": "C06"}],
"conditions": [{"key": "C01", "purpose": "governed-candidate-readback", "required_evidence": ["acceptance-proof"]}, {"key": "C06", "purpose": "governed-candidate-readback", "required_evidence": ["acceptance-proof"]}],
},
},
"mission-test",
+2 -1
View File
@@ -13,8 +13,9 @@ fixed clients must satisfy.
"""
import re
import pytest
from pathlib import Path
REPO = "C:/Users/ImRuY/aios-main-clone"
REPO = str(Path(__file__).resolve().parents[1]) # repo root (was hardcoded dev path)
ADMIN_HTML = f"{REPO}/active-runtime/app/static/admin-ai-evolution.html"
GITOPS_HTML = f"{REPO}/overlays/user-control-center/portal/gitops-dashboard/index.html"
+69 -5
View File
@@ -1,7 +1,30 @@
from pathlib import Path
import json, subprocess, sys
import json, os, subprocess, sys
def test_trusted_builder_emits_plan_bound_payload(tmp_path):
REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT = REPO_ROOT / "scripts/autonomous-release/build_trusted_release_artifact.py"
# ---------------------------------------------------------------------------
# RE-BIND REQUIRED after the governed true-merge of PR #88 lands on main.
# Set both constants to the exact merged main commit and its tree id, e.g.
# git -C <authority-repo> rev-parse refs/heads/main
# git -C <authority-repo> rev-parse 'refs/heads/main^{tree}'
# Until re-bound, the values below are all-zero placeholders: the canary then
# proves only that the successor-aware producer is wired, NOT that it is bound
# to the merged source. Do not treat an all-zero run as a live binding proof.
# ---------------------------------------------------------------------------
MERGED_SOURCE_OID = "0" * 40 # PLACEHOLDER - merged source commit (40 hex)
MERGED_TREE_OID = "0" * 40 # PLACEHOLDER - merged source tree oid (40 hex)
def _env_with_pythonpath() -> dict:
env = dict(os.environ)
root = str(REPO_ROOT)
env["PYTHONPATH"] = root + os.pathsep + env.get("PYTHONPATH", "")
return env
def _make_source(tmp_path) -> Path:
source = tmp_path / "source"; source.mkdir()
files = {
"active-runtime/app/deployment_state.py": "x\n",
@@ -11,13 +34,54 @@ def test_trusted_builder_emits_plan_bound_payload(tmp_path):
}
for rel, text in files.items():
path = source / rel; path.parent.mkdir(parents=True, exist_ok=True); path.write_text(text)
return source
def test_trusted_builder_emits_plan_bound_payload(tmp_path):
source = _make_source(tmp_path)
output = tmp_path / "release.tar"
script = Path(__file__).resolve().parents[1] / "scripts/autonomous-release/build_trusted_release_artifact.py"
result = subprocess.run([sys.executable, str(script), "--source", str(source), "--output", str(output),
result = subprocess.run([sys.executable, str(SCRIPT), "--source", str(source), "--output", str(output),
"--version", "V2R2C2", "--source-oid", "1" * 40, "--tree-oid", "2" * 40,
"--rollback-version", "V2R2C1", "--rollback-target-oid", "3" * 40,
"--rollback-artifact-sha256", "4" * 64, "--service", "aios-v2r2c2-production.service", "--port", "18928"],
check=True, capture_output=True, text=True)
check=True, capture_output=True, text=True, env=_env_with_pythonpath())
receipt = json.loads(result.stdout)
assert receipt["source_commit"] == "1" * 40
assert output.is_file() and receipt["artifact_size"] == output.stat().st_size
def test_successor_aware_binds_v2r2c3_live_binding_converge(tmp_path):
"""Successor-scoped fixture: the producer must bind the V2R2C3-LIVE-BINDING-CONVERGE
successor to its exact source_commit/tree_oid (no V2R2C2-only gate in source)."""
source = _make_source(tmp_path)
output = tmp_path / "v2r2c3.tar"
result = subprocess.run([sys.executable, str(SCRIPT), "--source", str(source), "--output", str(output),
"--version", "V2R2C3-LIVE-BINDING-CONVERGE",
"--source-oid", MERGED_SOURCE_OID,
"--tree-oid", MERGED_TREE_OID,
"--rollback-version", "V2R2C2",
"--rollback-target-oid", "5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
"--rollback-artifact-sha256", "43f12a14a248bea726dd9793587413c7e05d5c7cfd8704e1cabcc7d1d3214013",
"--service", "aios-gitops-model-api-v2r2c3-round3-release-control.service", "--port", "18933"],
check=True, capture_output=True, text=True, env=_env_with_pythonpath())
receipt = json.loads(result.stdout)
assert receipt["version"] == "V2R2C3-LIVE-BINDING-CONVERGE"
assert receipt["source_commit"] == MERGED_SOURCE_OID
assert receipt["tree_oid"] == MERGED_TREE_OID
assert output.is_file() and receipt["artifact_size"] == output.stat().st_size
def test_successor_aware_rejects_rollback_equal_version(tmp_path):
"""Fail-closed safety invariant preserved: rollback must never equal the payload version."""
source = _make_source(tmp_path)
for out in ("x.tar", "y.tar"):
broken = subprocess.run([sys.executable, str(SCRIPT), "--source", str(source), "--output", str(tmp_path / out),
"--version", "V2R2C3-LIVE-BINDING-CONVERGE",
"--source-oid", MERGED_SOURCE_OID,
"--tree-oid", MERGED_TREE_OID,
"--rollback-version", "V2R2C3-LIVE-BINDING-CONVERGE",
"--rollback-target-oid", "5cf20128d8be2d10aef87b09f8f7a2aecb2000f3",
"--rollback-artifact-sha256", "43f12a14a248bea726dd9793587413c7e05d5c7cfd8704e1cabcc7d1d3214013",
"--service", "x.service", "--port", "18933"],
capture_output=True, text=True, env=_env_with_pythonpath())
assert broken.returncode != 0 and "release_version_binding_invalid" in (broken.stderr + broken.stdout)