oats.core

Core configuration, profiles, tokens, and bus infrastructure.

oats.core.config

Configuration management with hierarchical loading.

Config priority (highest to lowest): 1. Environment variables set CODER_CONFIG_FILE to your coder.json file export CODER_CONFIG_FILE=./oats/config/coder.json’)

class oats.core.config.ProviderConfig(**data)

Bases: BaseModel

Configuration for an AI provider.

email: str | None

user_id: str | None

pw: str | None

api_key: str | None

base_url: str | None

enabled: bool

extra: dict[str, Any]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.core.config.ModelConfig(**data)

Bases: BaseModel

Configuration for model selection.

model_config: ClassVar[ConfigDict] = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

provider_id: str

model_id: str

class oats.core.config.HookEntry(**data)

Bases: BaseModel

A single hook configuration entry.

event: str

matcher: str | None

command: str

timeout: int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.core.config.HookConfig(**data)

Bases: BaseModel

Hook system configuration.

hooks: list[HookEntry]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.core.config.PermissionConfig(**data)

Bases: BaseModel

Permission system configuration.

read: dict[str, str]

write: dict[str, str]

bash: str

external_directory: dict[str, str]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oats.core.config.Config(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_prefix_target=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, _build_sources=None, **values)

Bases: BaseSettings

Main application configuration.

Loads from environment variables and config files.

provider: dict[str, ProviderConfig]

model: ModelConfig

permission: PermissionConfig

hooks: HookConfig

data_dir: Path

project_dir: Path

server_host: str

server_port: int

server_username: str | None

server_password: str | None

debug: bool

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'CODER_', 'env_prefix_target': 'variable', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod load(project_dir=None)

Load configuration from all sources.

Merges configs in priority order.

Return type:

Config

oats.core.config.get_config(project_dir=None, reload=False, verbose=False)

Get the global config instance, loading if necessary.

Return type:

Config

oats.core.config.get_data_dir()

Get the data directory, creating if needed.

Return type:

Path

oats.core.profiles

Feature profiles for coder — configurable packaging for different device targets.

Profiles let you run coder as slim and lightweight or as fully featured. Set via CODER_PROFILE env var or programmatically.

Built-in profiles:

• minimal — Core file/code tools only. No network, no browser, no cloud. Ideal for air-gapped devices, embedded, or CI runners.

• standard — Adds web search, S3, planning, memory, agents. Good default for most dev workstations.

• full — Everything enabled: browser/playwright, MCP, LSP, scraping

• custom — User controls each feature group via individual env vars.

Individual feature groups can always be overridden regardless of profile:

CODER_FEATURE_BROWSER=0    disables browser even on 'full'

class oats.core.profiles.FeatureProfile(name, core_tools=True, web_tools=True, s3_storage=False, database=False, redis=False, browser=False, scraping=False, mcp=False, lsp=False, planning=True, memory=True, agents=True)

Bases: object

Declares which feature groups are enabled for a profile.

name: str

core_tools: bool = True

web_tools: bool = True

s3_storage: bool = False

database: bool = False

redis: bool = False

browser: bool = False

scraping: bool = False

mcp: bool = False

lsp: bool = False

planning: bool = True

memory: bool = True

agents: bool = True

__init__(name, core_tools=True, web_tools=True, s3_storage=False, database=False, redis=False, browser=False, scraping=False, mcp=False, lsp=False, planning=True, memory=True, agents=True)

oats.core.profiles.get_profile()

Return the active feature profile (cached after first call).

Return type:

FeatureProfile

oats.core.profiles.is_feature_enabled(group)

Check if a feature group is enabled, with env override support.

Return type:

bool

Env override always wins:

CODER_FEATURE_BROWSER=1 → True (even if profile is ‘minimal’)

Otherwise falls back to the active profile.

oats.core.profiles.reset_profile()

Clear cached profile (for testing).

Return type:

None

oats.core.profiles.list_profiles()

Return the names of all built-in profiles.

Return type:

list[str]

oats.core.profiles.describe_profile(name=None)

Return a dict of group -> enabled for a profile (or the active one).

Return type:

dict[str, bool]

oats.core.tokens

Shared token estimation.

Used by the context ledger, session token budget, and conversation compactor so all three agree on what “tokens” means. Previously each site used len(text) // 4, which under-counts code (~2.5 chars/token) and is close to right for English prose (~3.5–4). The divergence matters when the budget decision (trigger compaction, elide sections) is based on these estimates: the real input payload routinely exceeds the approximation by 10–30%.

When tiktoken is importable we use cl100k_base as a stand-in — it’s not Claude’s exact BPE, but it’s a far better estimator than chars/4 for both prose and code and it’s stable across calls (so relative comparisons between sections are faithful).

When tiktoken is not available we fall back to len(text) / 3.5, which biases toward over-estimating. That’s the safer direction for budget decisions — we’d rather compact a turn early than blow the window.

The tokenizer (if any) is loaded lazily and memoized on the module so we don’t pay encoding-loader cost on every count_tokens() call.

oats.core.tokens.count_tokens(text)

Return an estimated token count for text.

Uses tiktoken’s cl100k_base when available, otherwise a conservative chars/3.5 approximation. Empty/None returns 0.

Return type:

int

oats.core.tokens.count_message_tokens(messages)

Sum token estimate across a list of session Message objects.

Accepts anything with get_text_content(), get_tool_calls(), and get_tool_results() so the existing SessionTokenBudget / compactor can call through without importing the Message type.

Return type:

int

oats.core.bus

Event bus for pub/sub pattern communication.

class oats.core.bus.EventType(*values)

Bases: str, Enum

Event types for the event bus.

SESSION_CREATED = 'session.created'

SESSION_UPDATED = 'session.updated'

SESSION_DELETED = 'session.deleted'

SESSION_BUDGET = 'session.budget'

SESSION_COMPACTED = 'session.compacted'

SESSION_TASK_BUDGET = 'session.task_budget'

MESSAGE_CREATED = 'message.created'

MESSAGE_UPDATED = 'message.updated'

MESSAGE_PART_CREATED = 'message.part.created'

MESSAGE_PART_UPDATED = 'message.part.updated'

TOOL_START = 'tool.start'

TOOL_COMPLETE = 'tool.complete'

TOOL_ERROR = 'tool.error'

PROVIDER_REQUEST = 'provider.request'

PROVIDER_RESPONSE = 'provider.response'

PROVIDER_ERROR = 'provider.error'

PERMISSION_REQUEST = 'permission.request'

PERMISSION_GRANTED = 'permission.granted'

PERMISSION_DENIED = 'permission.denied'

HOOK_FIRED = 'hook.fired'

HOOK_BLOCKED = 'hook.blocked'

FILE_CHANGED = 'file.changed'

AGENT_STARTED = 'agent.started'

AGENT_COMPLETED = 'agent.completed'

class oats.core.bus.Event(type, data=<factory>, source=None)

Bases: object

An event published to the bus.

type: EventType | str

data: dict[str, Any]

source: str | None = None

__init__(type, data=<factory>, source=None)

class oats.core.bus.EventBus

Bases: object

Simple pub/sub event bus for internal communication.

Supports both sync and async handlers.

__init__()

subscribe(event_type, handler)

Subscribe to events of a specific type.

Returns an unsubscribe function.

Return type:

Callable[[], None]

subscribe_all(handler)

Subscribe to all events.

Returns an unsubscribe function.

Return type:

Callable[[], None]

once(event_type, handler)

Subscribe to a single event of a specific type.

The handler will be automatically unsubscribed after being called once. Returns an unsubscribe function.

Return type:

Callable[[], None]

async publish(event)

Publish an event to all subscribers.

Return type:

None

publish_sync(event)

Publish an event synchronously (creates a new event loop if needed).

Return type:

None

clear()

Clear all handlers.

Return type:

None

oats.core.storage

File-based JSON storage system with locking.

exception oats.core.storage.StorageError

Bases: Exception

Base exception for storage errors.

class oats.core.storage.Storage(namespace, model_class)

Bases: Generic[T]

Generic file-based storage for Pydantic models.

Provides CRUD operations with file locking for concurrent access.

namespace: str

model_class: type[T]

property storage_dir: Path

Get the storage directory for this namespace.

async create(id, data)

Create a new item.

Return type:

TypeVar(T, bound= BaseModel)

async read(id)

Read an item by ID.

Return type:

Optional[TypeVar(T, bound= BaseModel)]

async update(id, data)

Update an existing item.

Return type:

TypeVar(T, bound= BaseModel)

async upsert(id, data)

Create or update an item.

Return type:

TypeVar(T, bound= BaseModel)

async delete(id)

Delete an item. Returns True if deleted, False if not found.

Return type:

bool

async list()

List all items.

Return type:

list[TypeVar(T, bound= BaseModel)]

async list_ids()

List all item IDs.

Return type:

list[str]

__init__(namespace, model_class)

class oats.core.storage.KeyValueStorage(name)

Bases: object

Simple key-value storage for arbitrary JSON data.

__init__(name)

property file_path: Path

Get the storage file path.

async get(key, default=None)

Get a value by key.

Return type:

Any

async set(key, value)

Set a value.

Return type:

None

async delete(key)

Delete a key. Returns True if deleted.

Return type:

bool

async all()

Get all data.

Return type:

dict[str, Any]

oats.core.features

Lightweight runtime feature flags for oats coder.

These are intentionally env-driven so new agent/runtime behaviors can be tested safely without requiring a larger rollout framework.

oats.core.features.deferred_tools_enabled()

Enable ToolSearch-style deferred tool loading.

Return type:

bool

oats.core.features.rich_compaction_enabled()

Enable denser engineering-oriented compaction prompts.

Return type:

bool

oats.core.features.streaming_tool_assembly_enabled()

Enable robust assembly of fragmented streamed tool calls.

Return type:

bool

oats.core.features.strict_tool_schemas_enabled()

Pass strict tool schema hints through to compatible providers.

Return type:

bool

oats.core.features.active_tool_guidance_enabled()

Explain active/deferred tools in the system prompt.

Return type:

bool

oats.core.features.reactive_compaction_candidate_enabled()

Enable prompt-too-long recovery by compacting the session and retrying the request instead of failing immediately.

Return type:

bool

oats.core.features.context_collapse_candidate_enabled()

Enable state-capsule style compaction that preserves summarized history as a system message for stronger continuation.

Return type:

bool

oats.core.features.lsp_tools_candidate_enabled()

Enable LSP-backed code-intelligence tools.

Return type:

bool

oats.core.features.token_budget_candidate_enabled()

Enable lightweight per-turn token budgeting and prompt guidance.

Return type:

bool

oats.core.features.task_budget_candidate_enabled()

Enable lightweight task-loop budgeting and anti-churn guidance.

Return type:

bool

oats.core.features.result_retention_enabled()

Enable compressed retention of large tool outputs in session history.

Return type:

bool

oats.core.features.trajectory_store_enabled()

Log every user prompt and tool outcome to the on-disk trajectory store.

Return type:

bool

oats.core.features.trajectory_retrieval_enabled()

Inject top-K past trajectories into the system prompt. Requires trajectory_store.

Return type:

bool

oats.core.features.cron_enabled()

Enable the coder cron/task scheduler tick in the interactive session.

Return type:

bool

oats.core.features.plugins_enabled()

Enable the declarative plugin loader (oats.plugins.loader.install).

Return type:

bool

oats.core.features.should_disable_streaming(model_id=None)

Check if streaming should be disabled for the current model.

Returns True if: - CODER_FEATURE_NON_STREAMING is explicitly enabled, OR - The model ID looks like Gemma 4 (auto-detection)

Auto-detection can be overridden with CODER_FEATURE_NON_STREAMING=0.

Return type:

bool

oats.core.id

ID generation utilities using uuid

oats.core.id.generate_id()

Generate a new unique ID.

Return type:

str

oats.core.id.generate_short_id()

Generate a shorter ID (first 8 chars).

Return type:

str

oats.core.offline

Offline-mode helpers.

Coder2’s core turn loop (provider → tools → session) does not phone home. All external traffic is either:

1. The LLM provider endpoint — user-configured; point at a local vLLM on localhost for a fully offline setup.

2. Explicit user-initiated actions via plugins (corpus fetchers, playwright, mattermost relays). These are surfaced in /help and the user has to invoke them directly.

When CODER_OFFLINE_STRICT=1 is set, anything that would reach outside localhost/127.0.0.1/.local/.internal has to consult require_network() first. The flag is advisory (we don’t patch the kernel namespace) — it gives plugin authors a single hook to check before opening a socket, and a single audit point users can grep for.

Typical usage inside a tool implementation:

from oats.core.offline import require_network
require_network(label="corpus:github", url=url)  # raises if blocked

exception oats.core.offline.OfflineBlockedError

Bases: RuntimeError

Raised when strict offline mode refuses an external request.

oats.core.offline.offline_strict()

True if CODER_OFFLINE_STRICT=1. Re-read per call so users can toggle it without restarting the session.

Return type:

bool

oats.core.offline.is_local_url(url)

Best-effort classification: does url target this machine or LAN?

Return type:

bool

oats.core.offline.require_network(*, label, url='')

Gate an outbound network operation on strict-offline mode.

Raises OfflineBlockedError when CODER_OFFLINE_STRICT=1 AND the target (url) is not local. Local targets (localhost, private-range IPs, .local) always pass. Plugins invoking explicit user actions can skip this — the point of strict mode is to gate implicit or accidental egress.

Return type:

None