Codumentor logo Codumentor

Best Practices

Guidelines for writing robust, performant, and maintainable Codumentor plugins.

Error Handling

Exceptions in handlers are caught by the bus; they do not crash the agent. Use logging:

from codumentor.utils.logging import get_logger

logger = get_logger("my_plugin")

async def my_handler(self, agent, ctx, payload):
    try:
        # Your logic
    except Exception as e:
        logger.error(f"Handler failed: {e}")
        return None  # Fail gracefully

Performance

Background Processing with Job Queue

For expensive operations (LLM calls, file processing), use the Job Queue System instead of direct async tasks:

from codumentor.core.job_queue import get_job_queue_registry, JobPriority

async def _on_response(self, agent, ctx, payload):
    registry = get_job_queue_registry(self.config)
    queue = registry.get_queue("background_llm")

    await queue.enqueue(
        job_type="my_job",
        args={"data": payload["assistant_text"]},  # Must be JSON-serializable
        priority=JobPriority.LOW,
    )

Benefits:

See Job Queue System for the full guide on registering job types, implementing gates, and handling recovered jobs.

Scheduled (recurring or one-off) work

When you need a job to run on a cron task — daily digests, periodic refresh, scheduled agent runs — use the Task engine. It's a thin layer over the job queue; you register a regular job_type handler and the task engine calls queue.enqueue(...) for you at the right times:

from codumentor.core.tasks import get_task_engine

task_engine = get_task_engine(self.config, registry)
await task_engine.create(
    owner_user_id=user_id,
    job_type="daily_summary",
    job_args={"repo": "team-repo"},
    cron_expression="0 9 * * 1-5",
    timezone="America/New_York",
)

See Task engine for firing semantics (advance-then-enqueue, overlap skip, catch-up cutoffs) and per-user queue routing.

Configuration

Plugins are configured in codumentor.yaml:

plugins:
  - module: "my_plugins.guardrail"
    class: "GuardrailPlugin"

  - module: "my_plugins.telemetry"
    class: "TelemetryPlugin"
    priority: 100  # Optional: override default priority
    args:
      log_level: "DEBUG"
      include_metrics: true

Configuration Fields

FieldRequiredDescription
moduleYesPython module path
classYesPlugin class name
argsNoArguments passed to constructor
priorityNoRegistration-priority override (lower = earlier)
dependenciesNoList of plugin names that must register before this one (overrides manifest/class dependencies). See Plugin Load Order.

Validating Configuration with Pydantic

Define a Pydantic model locally and build it from kwargs:

from pydantic import BaseModel

class MyPluginConfig(BaseModel):
    enabled: bool = True
    threshold: int = 100

class MyPlugin:
    def __init__(self, config=None, **kwargs):
        my_kwargs = {k: v for k, v in kwargs.items() if k in MyPluginConfig.model_fields}
        self.plugin_config = MyPluginConfig(**my_kwargs)

enabled: false does not stop your constructor

The loader instantiates every entry in plugins: — it never reads
enabled, and the host always calls register(). enabled is a convention
each plugin honours inside its own register(), so anything __init__ does
happens even for a plugin the operator switched off. A store built there
creates its directory, its SQLite file and the WAL/SHM sidecars for a disabled
plugin.

Keep constructors free of I/O and build the store on first read instead:

from functools import cached_property

class MyPlugin:
    def __init__(self, config=None, **kwargs):
        self.cfg = MyPluginConfig(**kwargs)
        self._enabled = self.cfg.enabled   # values only — no I/O

    @cached_property
    def _store(self) -> MyStore:
        return MyStore()

    async def register(self, bus):
        if not self._enabled:
            return
        self._store          # materialise it here, behind the guard
        bus.on("onContextReady", self._on_context_ready)

register() runs immediately after construction, so an enabled plugin still
opens its database at load time — an unwritable data directory stays a startup
failure. Built-in plugins spell this @lazy_store / warm_stores(self,
"_store") (codumentor/plugins/lazy_store.py); the tier-1 sweep
tests/test_plugin_constructor_no_writes.py fails any plugin in the tree whose
constructor writes to disk.

Making the enabled switch readable

Your own register() is not the only reader. The host asks too — it will not
mount a switched-off plugin's API router, and a sibling
looking for a collaborator (sibling_plugin()) skips one. All of them go
through plugin_bus.plugin_active(), which resolves the flag in this order:

  1. a public self.enabled,
  2. self._enabled — the spelling above, and the majority in the tree,
  3. enabled on whatever attribute holds an instance of the config_args_model() you declared (self.cfg for most plugins).

So the example above is enough, and so is doing nothing at all beyond parsing
your args into your declared model. What is not enough is folding the flag
into a derived object — a config the plugin builds from args plus
environment (Slack and Telegram do) is not the declared args model, so those
plugins publish self._enabled explicitly.

Get it wrong and nothing raises: a disabled plugin reads as a live
collaborator, which fails open. tests/test_plugin_enabled_contract.py
sweeps the whole roster for exactly this — any plugin whose args model declares
enabled must report it through plugin_active().

Large Content & Chunking

Tool results that exceed the chunk limit (controlled by tools.tool_result_max_content_length, default 30 000 characters) are automatically stored in an in-memory ContentStore and truncated, with a continuation hint appended. The agent can then page through the rest using the read_content tool. No plugin code is required for this to work.

This happens transparently in Tool.call(), so any tool that returns a large ToolResult.content is handled.

When to use the manual API

If your tool has access to the full content before it builds the ToolResult (e.g., it pre-truncates internally), you can store the content explicitly for better control:

async def execute(self, arguments):
    ctx = getattr(self, "_current_ctx", None) or {}
    content_store = ctx.get("content_store")

    full_text = extract_document(...)  # e.g. 85K chars

    if content_store:
        from codumentor.utils.content_store import ContentStore
        first_chunk, content_id, total_length, was_stored = (
            content_store.store_if_long(
                full_text,
                metadata={"source": self.name, "file": path},
            )
        )
        if was_stored:
            hint = ContentStore.continuation_hint(
                content_id, len(first_chunk), total_length,
            )
            return ToolResult(..., content=f"{header}\n{hint}\n\n{first_chunk}")

    return ToolResult(..., content=f"{header}\n\n{full_text}")

This is useful when:

Testing

Write unit tests using Python's unittest module:

import unittest
from my_plugin import MyPlugin

class TestMyPlugin(unittest.TestCase):
    def test_handler(self):
        plugin = MyPlugin()
        # Test handler logic...

A mock host makes session_state tests pass vacuously

session_state resolves its host by duck-typing: ctx["plugin_host"], then
agent.plugins, then agent.host, accepting whichever first has a
session_state attribute. A MagicMock agent has every attribute, so
mock.plugins.session_state(...) returns another mock, no store is ever
touched, and a test that then asserts on the mock passes while the real code
path is untested.

Give the test a real host:

from codumentor.plugins.plugin_bus import PluginHost

host = PluginHost()
ctx = {"plugin_host": host, "session_id": "s1", "kv": {}}
await plugin._on_prompt_assemble(agent, ctx, payload)

assert host.peek_session_state(plugin, "s1", "open_docs") == {...}

Assert through host.peek_session_state(...) (never through the mock), and
assert the release too: emit onSessionEvict on host.bus with
{"session_id": "s1"} and check peek_session_state returns None. The same
applies to agent.full_config and agent._source_ctx — a MagicMock answers
those as well, so use SimpleNamespace or a real object when the code branches
on them.

Two more mock-shaped traps, both from the shared host:

UI Component Guidelines

  1. Always escape user content: Use escapeHtml() for any dynamic text
  2. Support dark mode: Use CSS custom properties, not hardcoded colors
  3. Use Shadow DOM: Provides style isolation from the host
  4. Handle all states: Render appropriately for pending, success, error
  5. Keep bundles small: The SDK is ~15KB; aim for <50KB total per plugin

Prompt Templates

If your plugin contributes a subagent or any other LLM-driven tool, do not hard-code the system prompt as a Python string. Codumentor ships a Jinja2-based prompt system that every core agent and every first-party subagent already uses. Reusing it means users can override your prompt via --prompt-dir without forking your plugin.

from codumentor.utils.template import get_template_loader

loader = get_template_loader(self.full_config.prompt.template_dir)
system_prompt = loader.render_template(
    "my_plugin_system_prompt.jinja2",
    {"query": user_query, "repo_name": repo.name},
)

Drop the .jinja2 file into the repo's prompts/ directory (same place as agentic_developer_system_prompt.jinja2 and friends), or ship it alongside your plugin and document the filename so users can override it. {% include "general_instructions.jinja2" %} at the bottom of a subagent prompt will pull in Codumentor's shared tool-calling and environment rules for free.

See the Prompt System reference for the full file catalog, available variables, and override mechanics.

Example Plugins

These existing plugins serve as useful references:

See Also