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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions modules/communication/moltbot_bridge/ModLog.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,16 @@
# ModLog - moltbot_bridge

## 2026-07-18: REDDOG_VALVE_HIGH_AUTHORITY_CLASSIFICATION_PHASE1

**WSP Protocol**: WSP 00, 15, 22, 71, 97

- Classified `VALVE_OPEN_WORKTREE_CREATE` and `VALVE_OPEN_LIVE_ENQUEUE`
intent as HIGH authority even when `requested_operation` is otherwise LOW.
- Normalized the effective default valve state before classification and
serialization so empty input cannot emit an unclassified WORKTREE grant.
- Applied the classification at authority-profile seed, source, and delegated-
authority signing boundaries. This fix issues no authority or work itself.

## 2026-07-18: REDDOG_RESIDENT_LIVE_CANARY_PHASE1

**WSP Protocol**: WSP 00, 06, 15, 22, 46, 62, 71, 97
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@

from modules.communication.moltbot_bridge.src.reddog_signer_delegated_authority_runtime import (
HIGH_AUTHORITY_OPERATIONS,
HIGH_AUTHORITY_VALVE_STATES,
PrincipalAuthorityRecord,
)
from modules.communication.moltbot_bridge.src.reddog_work_order_signature_verifier import (
Expand Down Expand Up @@ -187,7 +188,12 @@ def run_reddog_authority_profile_seed_supply(
if not fid or not allow or not deny or not all(_path_within_foundup(path, fid) for path in (*allow, *deny)):
reasons.append(AuthorityProfileSeedSupplyReason.PATH_SCOPE_INVALID)
operation = str(requested_operation or "").strip()
if operation in HIGH_AUTHORITY_OPERATIONS and not (
effective_valve_state = str(valve_state_required or VALVE_OPEN_WORKTREE_CREATE)
high_authority = (
operation in HIGH_AUTHORITY_OPERATIONS
or effective_valve_state in HIGH_AUTHORITY_VALVE_STATES
)
if high_authority and not (
consensus_receipt_digest and sovereign_authorization_digest
):
reasons.append(AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING)
Expand Down Expand Up @@ -221,7 +227,7 @@ def run_reddog_authority_profile_seed_supply(
requested_operation=operation,
allowed_paths=allow,
denied_paths=deny,
valve_state_required=str(valve_state_required or VALVE_OPEN_WORKTREE_CREATE),
valve_state_required=effective_valve_state,
key_epoch=str(key_epoch or "epoch-1"),
required_tests=tuple(_strings(required_tests)) or _DEFAULT_REQUIRED_TESTS,
required_policy_gates=tuple(_strings(required_policy_gates)) or _DEFAULT_REQUIRED_POLICY_GATES,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

from modules.communication.moltbot_bridge.src.reddog_signer_delegated_authority_runtime import (
HIGH_AUTHORITY_OPERATIONS,
HIGH_AUTHORITY_VALVE_STATES,
PrincipalAuthorityRecord,
)
from modules.communication.moltbot_bridge.src.reddog_work_order_signature_verifier import (
Expand Down Expand Up @@ -223,7 +224,11 @@ def _seed_policy_reasons(seed: Mapping[str, Any], *, now_epoch: int, leeway_s: i
if not denied_paths or not all(_path_within_foundup(path, foundup_id) for path in denied_paths):
reasons.append(AuthorityProfileSourceSupplyReason.PATH_SCOPE)
operation = str(seed.get("requested_operation") or "")
if operation in HIGH_AUTHORITY_OPERATIONS and not (
high_authority = (
operation in HIGH_AUTHORITY_OPERATIONS
or str(seed.get("valve_state_required") or "") in HIGH_AUTHORITY_VALVE_STATES
)
if high_authority and not (
seed.get("consensus_receipt_digest") and seed.get("sovereign_authorization_digest")
):
reasons.append(AuthorityProfileSourceSupplyReason.HIGH_AUTHORITY_COSIGN)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@
"force_push",
}
)
HIGH_AUTHORITY_VALVE_STATES = frozenset(
{"VALVE_OPEN_LIVE_ENQUEUE", "VALVE_OPEN_WORKTREE_CREATE"}
)

_FOUNDUP_PATH_PREFIX = "modules/foundups/"

Expand Down Expand Up @@ -562,6 +565,7 @@ def issue_delegated_authority_runtime(
authority_tier = (
HIGH_AUTHORITY_TIER
if request.requested_operation in HIGH_AUTHORITY_OPERATIONS
or request.valve_state_required in HIGH_AUTHORITY_VALVE_STATES
else LOW_AUTHORITY_TIER
)
if authority_tier == HIGH_AUTHORITY_TIER and not (
Expand Down Expand Up @@ -742,6 +746,7 @@ def issue_delegated_authority_runtime(
"FailClosedPrincipalAuthorityResolver",
"FailClosedSignerClient",
"HIGH_AUTHORITY_OPERATIONS",
"HIGH_AUTHORITY_VALVE_STATES",
"InMemoryAuthorityRuntimeStore",
"IsolatedSignerClient",
"PrincipalAuthorityRecord",
Expand Down
8 changes: 8 additions & 0 deletions modules/communication/moltbot_bridge/tests/TestModLog.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,11 @@
## 2026-07-18: REDDOG_VALVE_HIGH_AUTHORITY_CLASSIFICATION_PHASE1

- Added signer, authority-seed, and authority-source regressions proving both
WORKTREE and LIVE_ENQUEUE intent require consensus plus sovereign evidence.
- Added LOW-operation, consensus-only, and empty/None default normalization
adversaries while retaining a positive LOW dry-run-only issuance case.
- Focused result: 45 tests passed across the three authority suites.

## 2026-07-18: REDDOG_RESIDENT_LIVE_CANARY_PHASE1

**Files**: `test_reddog_resident_live_canary.py` (NEW)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,11 @@ def _supply(tmp_path: Path, **overrides):


def test_seed_supply_writes_seed_consumable_by_source_supplier_and_promotion(tmp_path: Path) -> None:
seed_result = _supply(tmp_path)
seed_result = _supply(
tmp_path,
consensus_receipt_digest="sha256:consensus",
sovereign_authorization_digest="sha256:sovereign",
)

assert seed_result.accepted is True
assert seed_result.status == AUTHORITY_PROFILE_SEED_SUPPLY_ACCEPT
Expand Down Expand Up @@ -124,6 +128,42 @@ def test_seed_supply_rejects_high_authority_without_cosign(tmp_path: Path) -> No
assert AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING in result.rejection_reasons


def test_seed_supply_rejects_worktree_intent_for_low_operation_without_cosign(tmp_path: Path) -> None:
result = _supply(tmp_path, requested_operation="inspect_repo")

assert result.accepted is False
assert AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING in result.rejection_reasons


def test_seed_supply_rejects_live_enqueue_intent_without_cosign(tmp_path: Path) -> None:
result = _supply(
tmp_path,
requested_operation="inspect_repo",
valve_state_required="VALVE_OPEN_LIVE_ENQUEUE",
)

assert result.accepted is False
assert AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING in result.rejection_reasons


def test_seed_supply_normalizes_empty_worktree_intent_before_classification(tmp_path: Path) -> None:
for value in (None, ""):
result = _supply(
tmp_path,
requested_operation="inspect_repo",
valve_state_required=value,
)
assert result.accepted is False
assert AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING in result.rejection_reasons


def test_seed_supply_rejects_consensus_without_sovereign_authorization(tmp_path: Path) -> None:
result = _supply(tmp_path, consensus_receipt_digest="sha256:consensus")

assert result.accepted is False
assert AuthorityProfileSeedSupplyReason.HIGH_AUTHORITY_COSIGN_MISSING in result.rejection_reasons


def test_seed_supply_accepts_high_authority_with_cosign(tmp_path: Path) -> None:
result = _supply(
tmp_path,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,62 @@ def test_supplier_rejects_high_authority_without_cosign(tmp_path: Path) -> None:
assert AuthorityProfileSourceSupplyReason.HIGH_AUTHORITY_COSIGN in result.rejection_reasons


def test_supplier_rejects_worktree_intent_for_low_operation_without_cosign(tmp_path: Path) -> None:
seed = _seed(requested_operation="inspect_repo")
seed.pop("consensus_receipt_digest")
seed.pop("sovereign_authorization_digest")

result = run_reddog_authority_profile_source_artifact_supply(
repo_root=REPO_ROOT,
authority_seed=seed,
principal_authority_record=_principal(),
permission_snapshot=_snapshot(),
output_path=tmp_path / "runtime" / "authority_profile_source.json",
now_epoch=NOW,
)

assert result.accepted is False
assert AuthorityProfileSourceSupplyReason.HIGH_AUTHORITY_COSIGN in result.rejection_reasons


def test_supplier_rejects_live_enqueue_intent_without_cosign(tmp_path: Path) -> None:
seed = _seed(
requested_operation="inspect_repo",
valve_state_required="VALVE_OPEN_LIVE_ENQUEUE",
)
seed.pop("consensus_receipt_digest")
seed.pop("sovereign_authorization_digest")

result = run_reddog_authority_profile_source_artifact_supply(
repo_root=REPO_ROOT,
authority_seed=seed,
principal_authority_record=_principal(),
permission_snapshot=_snapshot(),
output_path=tmp_path / "runtime" / "authority_profile_source.json",
now_epoch=NOW,
)

assert result.accepted is False
assert AuthorityProfileSourceSupplyReason.HIGH_AUTHORITY_COSIGN in result.rejection_reasons


def test_supplier_rejects_consensus_without_sovereign_authorization(tmp_path: Path) -> None:
seed = _seed()
seed.pop("sovereign_authorization_digest")

result = run_reddog_authority_profile_source_artifact_supply(
repo_root=REPO_ROOT,
authority_seed=seed,
principal_authority_record=_principal(),
permission_snapshot=_snapshot(),
output_path=tmp_path / "runtime" / "authority_profile_source.json",
now_epoch=NOW,
)

assert result.accepted is False
assert AuthorityProfileSourceSupplyReason.HIGH_AUTHORITY_COSIGN in result.rejection_reasons


def test_supplier_rejects_holoindex_gap_authority(tmp_path: Path) -> None:
evidence = dict(_seed()["holoindex_evidence"])
evidence["index_gap_detected"] = True
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -314,9 +314,37 @@ def test_high_authority_requires_consensus_and_sovereign_authorization() -> None
assert RuntimeRejectCode.HIGH_AUTHORITY_NEEDS_COSIGN in result.receipt.rejection_reasons


def test_worktree_valve_intent_is_high_authority_even_for_low_operation() -> None:
result, _, _, _ = _issue(
requested_operation="inspect_repo",
consensus_receipt_digest=None,
sovereign_authorization_digest=None,
)
assert result.accepted is False
assert RuntimeRejectCode.HIGH_AUTHORITY_NEEDS_COSIGN in result.receipt.rejection_reasons


def test_live_enqueue_valve_intent_is_high_authority_even_for_low_operation() -> None:
result, _, _, _ = _issue(
requested_operation="inspect_repo",
valve_state_required="VALVE_OPEN_LIVE_ENQUEUE",
consensus_receipt_digest=None,
sovereign_authorization_digest=None,
)
assert result.accepted is False
assert RuntimeRejectCode.HIGH_AUTHORITY_NEEDS_COSIGN in result.receipt.rejection_reasons


def test_high_authority_rejects_consensus_without_sovereign_authorization() -> None:
result, _, _, _ = _issue(sovereign_authorization_digest=None)
assert result.accepted is False
assert RuntimeRejectCode.HIGH_AUTHORITY_NEEDS_COSIGN in result.receipt.rejection_reasons


def test_low_authority_can_issue_without_cosign() -> None:
result, _, _, _ = _issue(
requested_operation="inspect_repo",
valve_state_required="VALVE_OPEN_DRYRUN_ONLY",
consensus_receipt_digest=None,
sovereign_authorization_digest=None,
)
Expand Down
Loading