`vacant.composite`¶

P5 composite — composite parents, child manifests, Tree-Only protocol for sealed children, graduation (visibility flag flip — same keypair, same logbook), and the 3-layer collusion gate that graduation must pass.

manifest ¶

ChildManifest -- the dual-signed link between a composite parent and one of its children (P5 §3.1, dispatch §1).

The manifest is the only in-tree authorisation a child needs:

the parent vouches that this child is its child (signature_parent),
the child vouches that it accepts this parent (signature_child).

Both signatures cover the canonical-json of the same payload (D012 §C). A manifest with one signature missing or with either signature invalid is rejected by verify_or_raise.

The manifest is held by the composite parent's runtime (CompositeRuntime); it is not serialised to the public registry, so externally a composite vacant exposes only its own halo (P5 §3.3 black-box principle).

ChildManifest ¶

Bases: BaseModel

Dual-signed parent <-> child link.

closed_by_default is True for every D2 subagent-bud spawn (the path designed for composite children). For D1/D3/D5 spawns the manifest can still be issued, but the orchestrator typically reserves them for follow-on reasoning, not the canonical "closed sub-vacant" role.

signing_dict ¶

signing_dict() -> dict[str, Any]

Canonical dict over which both parent and child sign.

Excludes the two signature fields. Tool-whitelist lists are sorted so ["a","b"] and ["b","a"] produce the same payload.

Source code in src/vacant/composite/manifest.py

def signing_dict(self) -> dict[str, Any]:
    """Canonical dict over which both parent and child sign.

    Excludes the two signature fields. Tool-whitelist lists are
    sorted so `["a","b"]` and `["b","a"]` produce the same payload."""
    return {
        "parent_id": self.parent_id.hex(),
        "child_id": self.child_id.hex(),
        "birth_path": self.birth_path,
        "closed_by_default": self.closed_by_default,
        "tool_whitelist_inherited": sorted(self.tool_whitelist_inherited),
        "tool_whitelist_added": sorted(self.tool_whitelist_added),
        "tool_whitelist_removed": sorted(self.tool_whitelist_removed),
    }

verify ¶

verify() -> bool

True iff both signatures verify under their respective keys.

Source code in src/vacant/composite/manifest.py

def verify(self) -> bool:
    """True iff *both* signatures verify under their respective keys."""
    if not self.signature_parent or not self.signature_child:
        return False
    payload = self.signing_payload()
    if not verify(self.parent_id.verify_key(), payload, self.signature_parent):
        return False
    if not verify(self.child_id.verify_key(), payload, self.signature_child):
        return False
    return True

ensure_birth_path ¶

ensure_birth_path(value: str) -> Literal['D1', 'D2', 'D3', 'D4', 'D5']

Type-narrowing helper for callers building manifests from string inputs (e.g. logbook payloads).

Source code in src/vacant/composite/manifest.py

def ensure_birth_path(value: str) -> Literal["D1", "D2", "D3", "D4", "D5"]:
    """Type-narrowing helper for callers building manifests from string
    inputs (e.g. logbook payloads)."""
    if value not in BIRTH_PATHS:
        raise ManifestError(f"invalid birth_path {value!r}; expected one of {BIRTH_PATHS}")
    return value  # type: ignore[return-value]

tree_only ¶

Tree-Only outbound filter (P5 §2 D2 / dispatch §3).

A closed child can only call: - its parent (the composite root), or - a sibling within the same composite tree (mediated through the parent).

Cross-tree calls and direct external calls are rejected. The filter is the structural enforcement of D1 (Closed children is a hard principle): ACLs were rejected as too easy to bypass.

The orchestrator (CompositeRuntime) holds a list of ChildManifests. The tree itself is parent_id -> {child_id, ...}. is_call_allowed takes the caller's manifest and a callee identity and returns True iff the callee is the parent or a sibling registered under the same parent.

In production, every outbound socket from a closed-child runtime is gated through tree_only_filter middleware -- given the (caller_id, callee_id) pair, it raises TreeOnlyViolationError for non-tree calls.

siblings_of ¶

siblings_of(child_id: VacantId, manifests: Iterable[ChildManifest]) -> set[VacantId]

Return the set of other children sharing the same parent.

Excludes the child itself.

Source code in src/vacant/composite/tree_only.py

def siblings_of(child_id: VacantId, manifests: Iterable[ChildManifest]) -> set[VacantId]:
    """Return the set of *other* children sharing the same parent.

    Excludes the child itself."""
    by_parent: dict[VacantId, set[VacantId]] = {}
    for m in manifests:
        by_parent.setdefault(m.parent_id, set()).add(m.child_id)
    parent: VacantId | None = None
    for m in manifests:
        if m.child_id == child_id:
            parent = m.parent_id
            break
    if parent is None:
        return set()
    return by_parent.get(parent, set()) - {child_id}

is_call_allowed ¶

is_call_allowed(*, caller_manifest: ChildManifest, callee_id: VacantId, siblings: Iterable[VacantId] | None = None) -> bool

True iff a closed child may call callee_id.

Permitted callees: - caller_manifest.parent_id (the composite root). - any id in siblings (the orchestrator passes the precomputed sibling set; pass None to disable sibling calls entirely).

A child whose closed_by_default=False is treated as graduated -- it can reach any callee, so this filter no-ops and returns True.

Source code in src/vacant/composite/tree_only.py

def is_call_allowed(
    *,
    caller_manifest: ChildManifest,
    callee_id: VacantId,
    siblings: Iterable[VacantId] | None = None,
) -> bool:
    """True iff a closed child may call `callee_id`.

    Permitted callees:
    - `caller_manifest.parent_id` (the composite root).
    - any id in `siblings` (the orchestrator passes the precomputed
      sibling set; pass `None` to disable sibling calls entirely).

    A child whose `closed_by_default=False` is treated as graduated
    -- it can reach any callee, so this filter no-ops and returns True.
    """
    if not caller_manifest.closed_by_default:
        return True
    if callee_id == caller_manifest.parent_id:
        return True
    if siblings is not None and callee_id in set(siblings):
        return True
    return False

tree_only_filter ¶

tree_only_filter(*, caller_manifest: ChildManifest, callee_id: VacantId, siblings: Iterable[VacantId] | None = None) -> None

Raise TreeOnlyViolationError if the call would breach Tree-Only.

The orchestrator's outbound dispatch wires this in front of every network call from a closed-child runtime. A graduated child (closed_by_default=False) is unaffected.

Source code in src/vacant/composite/tree_only.py

def tree_only_filter(
    *,
    caller_manifest: ChildManifest,
    callee_id: VacantId,
    siblings: Iterable[VacantId] | None = None,
) -> None:
    """Raise `TreeOnlyViolationError` if the call would breach Tree-Only.

    The orchestrator's outbound dispatch wires this in front of every
    network call from a closed-child runtime. A graduated child
    (`closed_by_default=False`) is unaffected."""
    if not is_call_allowed(
        caller_manifest=caller_manifest,
        callee_id=callee_id,
        siblings=siblings,
    ):
        raise TreeOnlyViolationError(
            f"closed child {caller_manifest.child_id.short()} cannot call "
            f"{callee_id.short()} (not parent / not sibling)"
        )

graduation ¶

Graduation -- flip a closed child's registry_visibility from NONE to PUBLIC (P5 §3.7, dispatch §4).

Three preconditions, all must hold:

Parent consent -- a GraduationRequest signed by both parent and child (mirrors ChildManifest's dual-signature design).
Rate limit -- per-parent 24h sliding window, defaults to GRADUATION_RATE_LIMIT_PER_PARENT_24H (D012 §A).
Collusion check -- max signal strength on (parent, child) below GRADUATION_COLLUSION_THRESHOLD (D012 §B). Uses the CollusionDetector injected at construction time; a default CompositeStubDetector returning zeroes is used when P3 is not wired.

Identity preservation is load-bearing (CLAUDE.md §Closed children + graduation): the same keypair, the same logbook, just a visibility flag flip. The graduated child gets a new dual-signed manifest with closed_by_default=False and fresh GRADUATED log entries appended to both logbooks; nothing else changes.

A successful graduate() call returns a GraduationOutcome carrying the new manifest plus the CapabilityCard ready for publish_halo in P4. The caller is responsible for the actual halo publish (P5 does not import P4 directly to keep the layering clean).

GRADUATED_KIND `module-attribute` ¶

GRADUATED_KIND = 'COMPOSITE_GRADUATED'

Logbook entry kind written to both parent and child on success.

GraduationRequest `dataclass` ¶

GraduationRequest(parent_id: VacantId, child_id: VacantId, capability_text: str, parent_signature: bytes, child_signature: bytes)

Dual-signed authorisation for graduation.

parent_signature and child_signature are detached Ed25519 signatures over _signing_payload(parent_id, child_id, capability_text). Both must verify under their respective ids.

GraduationOutcome `dataclass` ¶

GraduationOutcome(new_manifest: ChildManifest, child_card: CapabilityCard, collusion_signals: CollusionSignals)

Result of a successful graduate().

new_manifest replaces the child's old (closed) manifest in the composite runtime. child_card is the freshly-signed capability card for publish_halo (P4); the caller actually publishes.

GraduationService ¶

GraduationService(*, rate_limit_per_24h: int = GRADUATION_RATE_LIMIT_PER_PARENT_24H, collusion_threshold: float = GRADUATION_COLLUSION_THRESHOLD, detector: CollusionDetector | None = None, clock: Callable[[], float] = time)

Stateful graduation gate held by the composite runtime.

Owns the per-parent rate-limit sliding-window deque + the injected collusion detector. The same instance is reused across graduation calls so the window persists.

The service does NOT publish to P4. It returns the CapabilityCard and the new dual-signed manifest; the caller wires up publish_halo (P4 §publish_halo) so P5 is not coupled to the registry implementation.

Source code in src/vacant/composite/graduation.py

def __init__(
    self,
    *,
    rate_limit_per_24h: int = GRADUATION_RATE_LIMIT_PER_PARENT_24H,
    collusion_threshold: float = GRADUATION_COLLUSION_THRESHOLD,
    detector: CollusionDetector | None = None,
    clock: Callable[[], float] = time.time,
) -> None:
    if rate_limit_per_24h < 1:
        raise ValueError(f"rate_limit_per_24h must be >= 1; got {rate_limit_per_24h}")
    if not (0.0 <= collusion_threshold <= 1.0):
        raise ValueError(f"collusion_threshold must be in [0, 1]; got {collusion_threshold}")
    self._rate_limit = rate_limit_per_24h
    self._collusion_threshold = collusion_threshold
    self._detector: CollusionDetector = detector or default_detector()
    self._clock = clock
    self._window_per_parent: dict[VacantId, deque[float]] = {}

graduate `async` ¶

graduate(*, runtime: CompositeRuntime, request: GraduationRequest, substrate_spec: SubstrateSpec | None = None, ts: float | None = None) -> GraduationOutcome

Run all three checks and, on success, flip the child's manifest in runtime and return the new outcome.

Raises GraduationConsentError, GraduationRateLimitError, or GraduationCollusionError on the first failing check. The runtime is not mutated until all checks pass.

Source code in src/vacant/composite/graduation.py

async def graduate(
    self,
    *,
    runtime: CompositeRuntime,
    request: GraduationRequest,
    substrate_spec: SubstrateSpec | None = None,
    ts: float | None = None,
) -> GraduationOutcome:
    """Run all three checks and, on success, flip the child's
    manifest in `runtime` and return the new outcome.

    Raises `GraduationConsentError`, `GraduationRateLimitError`, or
    `GraduationCollusionError` on the first failing check. The
    runtime is not mutated until all checks pass."""
    record = runtime.get_child(request.child_id)
    old_manifest = record.manifest
    if old_manifest.parent_id != request.parent_id:
        raise GraduationConsentError(
            f"graduation parent_id {request.parent_id.short()} does not match "
            f"child's manifest parent {old_manifest.parent_id.short()}"
        )
    if not request.verify():
        raise GraduationConsentError("graduation request signatures did not verify")

    when = ts if ts is not None else self._clock()
    self._enforce_rate_limit(request.parent_id, when)

    signals = self._detector.signals_for(request.parent_id, request.child_id)
    if max_signal_strength(signals) >= self._collusion_threshold:
        raise GraduationCollusionError(
            f"collusion signals too high for graduation: "
            f"controller={signals.same_controller:.2f}, "
            f"substrate={signals.same_substrate:.2f}, "
            f"stylo={signals.same_stylo:.2f} "
            f"(threshold {self._collusion_threshold})"
        )

    # All checks passed: rebuild manifest with closed_by_default=False,
    # mint capability card, append GRADUATED to both logbooks.
    new_manifest = self._rebuild_manifest(old_manifest)
    new_manifest = new_manifest.signed_by_parent(_parent_signing_key(runtime))
    new_manifest = new_manifest.signed_by_child(record.child_signing_key)

    spec = substrate_spec or record.child_form.substrate_spec
    child_card = CapabilityCard(
        vacant_id=request.child_id,
        capability_text=request.capability_text,
        substrate_spec=spec,
    ).signed(record.child_signing_key)

    runtime.mark_graduated(request.child_id, new_manifest)
    self._record_graduation(request.parent_id, when)
    _append_graduated_entries(
        runtime=runtime,
        request=request,
        child_form=record.child_form,
        child_signing_key=record.child_signing_key,
    )

    return GraduationOutcome(
        new_manifest=new_manifest,
        child_card=child_card,
        collusion_signals=signals,
    )

make_graduation_request ¶

make_graduation_request(*, parent_id: VacantId, parent_signing_key: SigningKey, child_id: VacantId, child_signing_key: SigningKey, capability_text: str) -> GraduationRequest

Builder helper: produce a fully signed GraduationRequest.

Source code in src/vacant/composite/graduation.py

def make_graduation_request(
    *,
    parent_id: VacantId,
    parent_signing_key: SigningKey,
    child_id: VacantId,
    child_signing_key: SigningKey,
    capability_text: str,
) -> GraduationRequest:
    """Builder helper: produce a fully signed `GraduationRequest`."""
    payload = _signing_payload(parent_id, child_id, capability_text)
    return GraduationRequest(
        parent_id=parent_id,
        child_id=child_id,
        capability_text=capability_text,
        parent_signature=sign(parent_signing_key, payload),
        child_signature=sign(child_signing_key, payload),
    )

collusion ¶

Collusion-detection adapter for the graduation flow (P5 §6, dispatch §5).

Before a closed child's halo is published, the graduation flow must confirm that the parent <-> child pair does NOT exhibit the canonical collusion signals (same-controller, same-substrate, same-stylo). Above-threshold signals indicate the parent is trying to graduate a sock-puppet rather than a genuinely independent capability.

This module wraps P3's same_detect for the graduation path. P3 is not a hard dependency: callers may pass any CollusionDetector Protocol implementation. A default_detector() is provided that returns 0.0 on every signal -- safe default so graduation works in P3-less builds and defers to other gates (parent consent + rate limit). Tests inject a detector that returns specific signal strengths.

Per the stated theory invariant (CLAUDE.md): same- detection raises cost, not prevents. The graduation gate's role is the same: a high collusion signal blocks this* graduation; the attacker can re-attempt after burning more identity capital.

CollusionSignals `dataclass` ¶

CollusionSignals(same_controller: float, same_substrate: float, same_stylo: float)

Strengths of the three collusion signals (THEORY_V5 §6 framing).

Each strength is in [0.0, 1.0]. same_controller -- both vacants appear to be operated by the same human/org. same_substrate -- both run on the same base model family. same_stylo -- the children's behavioural fingerprints (STYLO Vec16) are within the drift threshold of each other or of the parent.

CollusionDetector ¶

Bases: Protocol

Plug-point for the graduation flow.

Concrete impls call into P3's same_detect (substrate fingerprint comparison, controller graph, STYLO embedding distance). Tests pass a stub returning fixed signals.

CompositeStubDetector `dataclass` ¶

CompositeStubDetector(same_controller: float = 0.0, same_substrate: float = 0.0, same_stylo: float = 0.0)

Constant-returning detector. Useful in tests + the P3-less default.

max_signal_strength ¶

max_signal_strength(signals: CollusionSignals) -> float

Conservative composition: the highest signal is the trip line.

Source code in src/vacant/composite/collusion.py

def max_signal_strength(signals: CollusionSignals) -> float:
    """Conservative composition: the *highest* signal is the trip line."""
    return max(signals.same_controller, signals.same_substrate, signals.same_stylo)

default_detector ¶

default_detector() -> CollusionDetector

Return a no-signal detector. Used when P3 is not wired (D012 §B); graduation then defers to parent consent + rate limit.

Source code in src/vacant/composite/collusion.py

def default_detector() -> CollusionDetector:
    """Return a no-signal detector. Used when P3 is not wired (D012 §B);
    graduation then defers to parent consent + rate limit."""
    return CompositeStubDetector()

orchestrator ¶

Composite orchestrator (P5 §3, dispatch §2).

CompositeRuntime holds a composite parent's ResidentForm plus the ChildManifest list for every direct child, and exposes:

delegate(subtask, child_id) -- dispatch a subtask to a sub-vacant through the Tree-Only protocol.
aggregate(child_responses) -- combine sub-results into the composite's response.
outbound_call(caller_child_id, callee_id) -- the gate every outgoing socket from a closed-child runtime passes through.

Every dispatch writes to both logbooks: the parent's logbook records the delegation, the child's logbook records the inbound task. That dual-write is the audit trail used by graduation + by the internal mini_rep tracker (P5 §3.6).

ChildHandler `module-attribute` ¶

ChildHandler = Callable[[Any], Coroutine[Any, Any, Any]]

Async callable invoked when the orchestrator delegates to a child. The orchestrator hands the subtask in; the child returns its result. Real implementations would route via P6 envelopes; tests pass a lambda.

ChildRecord `dataclass` ¶

ChildRecord(manifest: ChildManifest, child_form: ResidentForm, child_signing_key: SigningKey, handler: ChildHandler)

A child registered with the composite parent.

Combines the dual-signed manifest with the runtime hooks the orchestrator needs to dispatch to it.

DelegationResult `dataclass` ¶

DelegationResult(child_id: VacantId, subtask: Any, response: T)

Output of CompositeRuntime.delegate.

CompositeRuntime ¶

CompositeRuntime(*, parent_form: ResidentForm, parent_signing_key: SigningKey)

In-process composite parent. Holds the parent form + child registry.

Construction does not perform any I/O. Use register_child to add a fully-spawned child + its dual-signed manifest. delegate dispatches a single subtask; aggregate combines a list of DelegationResults into one output.

Concurrency: delegation acquires self._lock per dispatch so that parent-logbook writes stay strictly ordered (BLAKE2b chain integrity). Children are dispatched sequentially in delegate_many by default; callers who want parallelism inject their own asyncio.gather through delegate.