Skip to content

symfonic.platform.resume_budget

resume_budget

Admission and raw usage proof for a budgeted human continuation.

ResumeBudgetAdmission dataclass 

ResumeBudgetAdmission(prepared_plan: Any, event_guard: ResumeUsageGuard)

One pinned compilation and one fresh stream-local usage observer.

ResumeUsageGuard 

ResumeUsageGuard()

Reject unmeasured rounds before the raw stream reaches its projection.

Source code in src/symfonic/platform/resume_budget.py 
20
21
def __init__(self) -> None:
    self._measured_rounds = 0

admit_resume_budget 

admit_resume_budget(agent: Any, issuer_scope: Any) -> ResumeBudgetAdmission

Validate the issuer composition before claiming its continuation token.

The caller decides whether a budget is configured. This helper always requires the human/telemetry envelope and never consults a global tracker.

Source code in src/symfonic/platform/resume_budget.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def admit_resume_budget(agent: Any, issuer_scope: Any) -> ResumeBudgetAdmission:
    """Validate the issuer composition before claiming its continuation token.

    The caller decides whether a budget is configured. This helper always
    requires the human/telemetry envelope and never consults a global tracker.
    """
    from symfonic.agent.facade import Agent
    from symfonic.capabilities.human.capability import HumanInteractionCapability
    from symfonic.core.observability.metrics import ConversationMetricsCollector
    from symfonic.core.observability.pricing import calculate_cost_with_meta
    from symfonic.platform.telemetry import TelemetryCapability, _tenant_of

    if type(agent) is not Agent or sorted(agent.capabilities) != ["human", "telemetry"]:
        raise _unavailable()
    owner = agent.event_sink_owner
    composed = agent._composed_capabilities
    if (
        len(composed) != 2
        or sum(type(capability) is HumanInteractionCapability for capability in composed) != 1
        or sum(type(capability) is TelemetryCapability for capability in composed) != 1
        or not any(capability is owner for capability in composed)
    ):
        raise _unavailable()
    tenant = _tenant_of(issuer_scope)
    if (
        type(owner) is not TelemetryCapability
        or not tenant
        or owner.tenant != tenant
        or type(owner._collector) is not ConversationMetricsCollector
        or not agent.stream_usage_declared
    ):
        raise _unavailable()
    resolved_config = agent._plans._resolve_config()
    model_name = resolved_config.model_name
    if not model_name or calculate_cost_with_meta(model_name, 0, 0)[1]:
        raise _unavailable()
    prepared_plan = agent._prepare("(resume)", None, resolved_config=resolved_config)
    if prepared_plan.model.model_name != model_name:
        raise _unavailable()
    return ResumeBudgetAdmission(prepared_plan, ResumeUsageGuard())