converge.policy

Admission, trust, governance, and safety. AdmissionPolicy implementations (Open, Whitelist, Token) control who can join a pool; pools use them when joining. TrustModel maintains per-agent scores and supports discovery trust thresholds. GovernanceModel implementations resolve disputes: Democratic (majority vote), Dictatorial (single leader), Bicameral (two chambers must agree), Veto (majority with a veto counterpower), Empirical (evidence-weighted votes). Safety provides ResourceLimits, ActionPolicy (allowlist of actions), and validate_safety for resource checks. The StandardExecutor (runtime) accepts an optional safety_policy (ResourceLimits, ActionPolicy) to enforce action allowlists and task resource limits. Pools can set trust_model and trust_threshold; PoolManager.join_pool rejects agents whose trust score is below the threshold.

Custom governance models

You can define your own governance by subclassing GovernanceModel and implementing resolve_dispute(context) -> Any:

1. Subclass converge.policy.governance.GovernanceModel.

2. Implement resolve_dispute(self, context: Any) -> Any. The context is whatever the caller provides when resolving a dispute (e.g. a dict with "votes", "weights", or custom keys). Return the chosen outcome, or None for no decision / deadlock.

3. Use the model either by passing it when creating a pool: pool_manager.create_pool({"id": "my-pool", "governance_model": my_model}), or by calling my_model.resolve_dispute(context) yourself when a dispute arises.

The pool stores the instance on pool.governance_model (if provided at creation). Built-in models (Democratic, Dictatorial, Bicameral, Veto, Empirical) document their expected context shape in their docstrings; your custom model can define any contract you need.

class converge.policy.admission.AdmissionPolicy

Bases: ABC

Abstract base class for defining rules on how agents are admitted to a pool.

abstractmethod can_admit(agent_id: str, pool_context: dict) → bool

Determine if an agent is authorized to join.

Parameters:

• agent_id (str) – The unique fingerprint of the agent requesting admission.

• pool_context (dict) – Arbitrary metadata about the pool or request (e.g., existing members, tokens).

Returns:

True if the agent is admitted, False otherwise.

Return type:

bool

class converge.policy.admission.OpenAdmission

Bases: AdmissionPolicy

A permissive policy that allows any agent to join.

can_admit(agent_id: str, pool_context: dict) → bool

Check admission. Always returns True.

Parameters:

• agent_id (str) – Agent fingerprint.

• pool_context (dict) – Pool context.

Returns:

True.

Return type:

bool

class converge.policy.admission.WhitelistAdmission(whitelist: list[str])

Bases: AdmissionPolicy

A restrictive policy that only allows agents present in a predefined whitelist.

__init__(whitelist: list[str])

Initialize the whitelist policy.

Parameters:

whitelist (List[str]) – A list of authorized agent IDs.

can_admit(agent_id: str, pool_context: dict) → bool

Check if the agent is in the whitelist.

Parameters:

• agent_id (str) – Agent fingerprint.

• pool_context (dict) – Pool context.

Returns:

True if whitelisted, False otherwise.

Return type:

bool

class converge.policy.admission.TokenAdmission(required_token: str)

Bases: AdmissionPolicy

A policy requiring a specific secret token to be present in the request context.

__init__(required_token: str)

Initialize with the required token.

Parameters:

required_token (str) – The secret token that must be provided.

can_admit(agent_id: str, pool_context: dict) → bool

Check if the correct token is provided in the pool context.

Parameters:

• agent_id (str) – Agent fingerprint.

• pool_context (dict) – Context containing the ‘token’ key.

Returns:

True if token matches, False otherwise.

Return type:

bool

class converge.policy.trust.TrustModel

Bases: object

Computes and manages trust scores for agents within the network.

Trust is calculated based on historical interactions, direct feedback, and configured policies.

update_trust(agent_id: str, score_delta: float) → float

Update the trust score for an agent.

Parameters:

• agent_id (str) – The agent ID.

• score_delta (float) – The change in score (positive or negative).

Returns:

The new trust score.

Return type:

float

get_trust(agent_id: str) → float

Retrieve the current trust score for an agent.

Parameters:

agent_id (str) – The agent ID.

Returns:

A score between 0.0 and 1.0.

Return type:

float

class converge.policy.governance.GovernanceModel

Bases: ABC

Abstract base class for pool governance models. Determines how decisions are made within a pool (e.g. voting, dictatorial).

Implementing a custom governance model

Subclass GovernanceModel and implement resolve_dispute(self, context: Any) -> Any. The context is supplied by the caller when resolving a dispute (e.g. a dict with "votes", "chamber_a_votes", or any structure your model expects). Return the chosen outcome, or None to indicate no decision / deadlock.

Example:

class QualifiedMajorityGovernance(GovernanceModel):
    def __init__(self, threshold: float = 0.66):
        self.threshold = threshold

    def resolve_dispute(self, context: Any) -> Any:
        votes = context.get("votes", []) if isinstance(context, dict) else []
        if not votes:
            return None
        from collections import Counter
        count = Counter(votes)
        top, n = count.most_common(1)[0]
        return top if n >= len(votes) * self.threshold else None

Pass an instance when creating a pool (e.g. create_pool({"id": "p1", "governance_model": my_model})) or call resolve_dispute(context) yourself when a dispute arises.

abstractmethod resolve_dispute(context: Any) → Any

Resolve a dispute or deadlock.

Parameters:

context – Caller-defined data (e.g. dict with “votes”, “weights”, or custom keys). Structure depends on the governance model.

Returns:

The chosen outcome, or None if no decision can be made.

class converge.policy.governance.DictatorialGovernance(leader_id: str)

Bases: GovernanceModel

A single leader makes all critical decisions.

resolve_dispute(context: Any) → Any

Return the leader’s decision.

class converge.policy.governance.DemocraticGovernance

Bases: GovernanceModel

Decisions are made by majority vote using Consensus.majority_vote. Expects context to contain a ‘votes’ key (list of vote options).

resolve_dispute(context: Any) → Any

Resolve by majority vote over context[‘votes’].

class converge.policy.governance.BicameralGovernance

Bases: GovernanceModel

Two chambers must agree (separated powers). Each chamber decides by majority; the final outcome is adopted only if both chambers choose the same option.

Context: dict with “chamber_a_votes” and “chamber_b_votes” (lists of options). Optional “tie_breaker” (any): if chambers disagree, return this instead of None.

resolve_dispute(context: Any) → Any

Resolve a dispute or deadlock.

Parameters:

context – Caller-defined data (e.g. dict with “votes”, “weights”, or custom keys). Structure depends on the governance model.

Returns:

The chosen outcome, or None if no decision can be made.

class converge.policy.governance.VetoGovernance(veto_agent_id: str | None = None)

Bases: GovernanceModel

Counterpower: a designated agent can veto the outcome of a majority vote. The body votes; if the veto is not exercised, the majority wins; otherwise the decision is blocked (returns None or a fallback).

Context: “votes” (list), “veto_exercised” (bool), optional “veto_agent_id”, optional “fallback” when veto is exercised.

resolve_dispute(context: Any) → Any

Resolve a dispute or deadlock.

Parameters:

context – Caller-defined data (e.g. dict with “votes”, “weights”, or custom keys). Structure depends on the governance model.

Returns:

The chosen outcome, or None if no decision can be made.

class converge.policy.governance.EmpiricalGovernance

Bases: GovernanceModel

Evidence-weighted decision: votes are combined with weights (e.g. confidence or evidence strength). The option with the highest total weight wins; ties yield None.

Context: “votes” (list of options), optional “weights” (list of float, same length as votes: weight per voter). If “weights” is absent, falls back to unweighted majority.

resolve_dispute(context: Any) → Any

Resolve a dispute or deadlock.

Parameters:

context – Caller-defined data (e.g. dict with “votes”, “weights”, or custom keys). Structure depends on the governance model.

Returns:

The chosen outcome, or None if no decision can be made.

class converge.policy.safety.ResourceLimits(max_cpu_tokens: float = 1.0, max_memory_mb: int = 512, max_network_requests: int = 100)

Bases: object

Defines upper bounds for resource consumption by an agent or task.

max_cpu_tokens

Maximum virtual CPU units allowed.

Type:

float

max_memory_mb

Maximum memory in megabytes.

Type:

int

max_network_requests

Maximum network calls per time window.

Type:

int

max_cpu_tokens: float = 1.0

max_memory_mb: int = 512

max_network_requests: int = 100

class converge.policy.safety.ActionPolicy(allowed_actions: list[str] | None = None)

Bases: object

Controls which actions an agent is permitted to execute.

If ‘allowed_actions’ is provided, only those actions are allowed (allowlist). If ‘allowed_actions’ is None, all actions are allowed (permissive).

__init__(allowed_actions: list[str] | None = None)

Initialize the action policy.

Parameters:

allowed_actions (List[str]) – List of allowed action names. If None, all actions are allowed (permissive).

allowed_actions: set[str] | None

is_allowed(action_name: str) → bool

Check if a specific action is authorized.

Parameters:

action_name (str) – The name of the action to check.

Returns:

True if allowed, False otherwise.

Return type:

bool

converge.policy.safety.validate_safety(limits: ResourceLimits, requested_cpu: float, requested_mem: int) → bool

Validate that a resource request is within the defined limits.

Parameters:

• limits (ResourceLimits) – The active resource constraints.

• requested_cpu (float) – The amount of CPU requested.

• requested_mem (int) – The amount of memory requested.

Returns:

True if within limits, False if exceeded.

Return type:

bool