user_gpt_config

Per-user policy for a user-linked ChatGPT/Codex quota backend.

The setting is deliberately separate from user_llm_config: the latter is a per-channel, user-supplied OpenAI-compatible endpoint, while !gpt selects the already-linked Codex account for one resolved Stargazer identity. No credential or external endpoint is ever stored in Redis.

exception user_gpt_config.GptCommandError

Bases: ValueError

A safe, user-facing !gpt grammar error.

exception user_gpt_config.GptRouteStoreUnavailable

Bases: RuntimeError

Route state could not be read safely, so provider fallback is forbidden.

class user_gpt_config.CodexQuotaAccess(account_ids, has_own_account)

Bases: object

Ordered account generations authorized for one beneficiary.

Parameters:
account_ids: tuple[str, ...]
has_own_account: bool
property account_id: str
class user_gpt_config.GptRoute(model_alias, effort, account_id=None, model_policy='fixed', effort_policy='fixed', model_candidates=(), effort_candidates=(), pair_candidates=(), streaming='default', speed='standard', max_length='default')

Bases: object

Validated stored GPT selection and delivery policy.

The first three fields retain the v1 constructor contract. In an automatic policy, model_alias and effort are deterministic fallbacks and the corresponding candidate tuple is the selector’s permitted domain.

Parameters:

  • model_alias (str)

  • effort (str)

  • account_id (str | None)

  • model_policy (Literal['fixed', 'auto'])

  • effort_policy (Literal['fixed', 'auto'])

  • model_candidates (tuple[str, ...])

  • effort_candidates (tuple[str, ...])

  • pair_candidates (tuple[tuple[str, str], ...])

  • streaming (Literal['default', 'on', 'off'])

  • speed (Literal['standard', 'fast'])

  • max_length (Literal['default', 'short', 'medium', 'long', 'heavy'])

model_alias: str
effort: str
account_id: str | None
model_policy: Literal['fixed', 'auto']
effort_policy: Literal['fixed', 'auto']
model_candidates: tuple[str, ...]
effort_candidates: tuple[str, ...]
pair_candidates: tuple[tuple[str, str], ...]
streaming: Literal['default', 'on', 'off']
speed: Literal['standard', 'fast']
max_length: Literal['default', 'short', 'medium', 'long', 'heavy']
property model_id: str
property effort_label: str
property allowed_models: tuple[str, ...]
property allowed_efforts: tuple[str, ...]
property allowed_pairs: tuple[tuple[str, str], ...]

Return only model/effort combinations validated for this policy.

property is_auto: bool
class user_gpt_config.GptResolution(model_alias, effort, account_id, streaming, speed, max_length, fallback_account_ids=())

Bases: object

One immutable, account-bound route resolved from a GptRoute.

Parameters:

  • model_alias (str)

  • effort (str)

  • account_id (str)

  • streaming (bool)

  • speed (Literal['standard', 'fast'])

  • max_length (Literal['default', 'short', 'medium', 'long', 'heavy'])

  • fallback_account_ids (tuple[str, ...])

model_alias: str
effort: str
account_id: str
streaming: bool
speed: Literal['standard', 'fast']
max_length: Literal['default', 'short', 'medium', 'long', 'heavy']
fallback_account_ids: tuple[str, ...]
property model_id: str
property effort_label: str
class user_gpt_config.GptSettingsPatch(streaming=None, speed=None, max_length=None)

Bases: object

A parsed, immutable partial update to delivery policy.

Parameters:

  • streaming (Literal['default', 'on', 'off'] | None)

  • speed (Literal['standard', 'fast'] | None)

  • max_length (Literal['default', 'short', 'medium', 'long', 'heavy'] | None)

streaming: Literal['default', 'on', 'off'] | None
speed: Literal['standard', 'fast'] | None
max_length: Literal['default', 'short', 'medium', 'long', 'heavy'] | None
apply(route)
Return type:

GptRoute

Parameters:

route (GptRoute)

class user_gpt_config.GptCommand(action, route=None, settings=None)

Bases: object

Parameters:
action: Literal['status', 'help', 'off', 'enable', 'settings']
route: GptRoute | None
settings: GptSettingsPatch | None
class user_gpt_config.CodexRouteTarget(account_id, effort, fallback_account_ids=())

Bases: object

Validated internal route pinned to one auth generation.

Parameters:

  • account_id (str)

  • effort (str)

  • fallback_account_ids (tuple[str, ...])

account_id: str
effort: str
fallback_account_ids: tuple[str, ...]
property account_ids: tuple[str, ...]
user_gpt_config.effort_label(canonical_effort)

Return the public command label for one canonical SDK effort.

Return type:

str

Parameters:

canonical_effort (str)

user_gpt_config.subject_for_message(msg)

Return the stable, global identity used by the per-user route key.

Return type:

str

Parameters:

msg (Any)

user_gpt_config.redis_key(subject)
Return type:

str

Parameters:

subject (str)

user_gpt_config.is_codex_account_id(value)

Return whether value is an opaque server-generated account id.

Return type:

bool

Parameters:

value (str | None)

user_gpt_config.codex_route_url(effort, account_id, fallback_account_ids=())

Build an internal route pinned to an ordered set of auth generations.

Return type:

str

Parameters:

  • effort (str)

  • account_id (str)

  • fallback_account_ids (tuple[str, ...])

user_gpt_config.codex_target_from_url(url)

Decode a trusted Codex route marker, or return None.

Return type:

CodexRouteTarget | None

Parameters:

url (str | None)

user_gpt_config.codex_effort_from_url(url)

Decode a trusted Codex route marker, or return None for HTTP routes.

Return type:

str | None

Parameters:

url (str | None)

user_gpt_config.codex_account_from_url(url)

Return the opaque account generation carried by an internal route.

Return type:

str | None

Parameters:

url (str | None)

user_gpt_config.codex_accounts_from_url(url)

Return every ordered auth generation carried by an internal route.

Return type:

tuple[str, ...]

Parameters:

url (str | None)

user_gpt_config.is_codex_route_url(url)
Return type:

bool

Parameters:

url (str | None)

user_gpt_config.is_codex_model_id(model)

Return whether model is one of the server-approved Codex aliases.

Return type:

bool

Parameters:

model (str | None)

user_gpt_config.parse_gpt_command(text)

Parse the public !gpt grammar without mutating state.

Return type:

GptCommand

Parameters:

text (str)

user_gpt_config.retain_gpt_route_pairs(route, pairs)

Return an automatic policy restricted to supported combinations.

Return type:

GptRoute

Parameters:
user_gpt_config.bind_gpt_route(route, account_id)

Return a policy pinned to one validated account generation.

Return type:

GptRoute

Parameters:
user_gpt_config.bind_gpt_resolution_accounts(resolution, account_ids)

Pin one resolved policy to its immutable, ordered account plan.

Return type:

GptResolution

Parameters:
user_gpt_config.resolve_gpt_route(route, *, model_alias=None, effort=None)

Resolve one policy without mutation, using its first candidates as fallback.

A caller may supply an automatic selector’s choices. Missing automatic choices deterministically use the first permitted candidate, so selector failure never changes provider or escapes the user’s policy.

Return type:

GptResolution

Parameters:
user_gpt_config.resolution_from_codex_transport(*, model_id, chat_url, service_tier, streaming, max_length)

Rebuild the effective public route after checkpoint restoration.

The executor may replace a newly constructed turn’s route with the first-writer route stored in its durable checkpoint. Presentation must follow that restored route too, rather than the user’s possibly newer !gpt policy.

Return type:

GptResolution

Parameters:

  • model_id (str)

  • chat_url (str)

  • service_tier (str | None)

  • streaming (bool)

  • max_length (str)

user_gpt_config.summarize_gpt_route(route)

Return a compact prose summary suitable for command responses.

Return type:

str

Parameters:

route (GptRoute)

async user_gpt_config.get_user_gpt_route(redis, subject)

Read a route from the primary Redis client and fail closed if malformed.

Return type:

GptRoute | None

Parameters:
async user_gpt_config.set_user_gpt_route(redis, subject, route, *, updated_by, expected_existing=None)

Store one v3 policy only while its exact quota source is authorized.

Return type:

str | None

Parameters:
async user_gpt_config.clear_user_gpt_route(redis, subject)

Disable the GPT route for exactly one canonical user.

Return type:

str | None

Parameters:
async user_gpt_config.get_codex_quota_access(redis, subject, *, routing_seed=None, frozen_account_ids=None)

Resolve self first, then generation-valid sponsored fallback accounts.

frozen_account_ids is the transport revalidation boundary for an already-routed turn. It verifies those exact generations against the full live grant set without applying a second, independently ordered 32-account cap. The current own account is still surfaced even when it is outside the frozen plan so transport can fail closed instead of skipping newly linked self quota.

Return type:

CodexQuotaAccess | None

Parameters:

  • redis (Any)

  • subject (str)

  • routing_seed (str | None)

  • frozen_account_ids (tuple[str, ...] | None)

async user_gpt_config.handle_gpt_command(processor, msg, platform)

Serialize and execute one identity-scoped !gpt command.

Return type:

bool

Parameters: