Backend Hooks

This is the complete reference for Codumentor plugin lifecycle hooks. Plugins intercept agent execution via hooks. Each hook handler receives:

agent: The Agent instance
ctx: Shared context dict (see Overview)
payload: Hook-specific data

Hook Reference

Lifecycle Hooks

Hook	Payload	Purpose
`onServerReady`	`{}`	Fired after the API server is fully initialized and accepting requests. Use for deferred, non-critical initialization that should not delay startup.
`onSessionEvict`	`{session_id}`	Fired when a session or cached agent is evicted from memory (LRU or explicit deletion). Use to release session-scoped resources such as connection pools, file handles, or external subscriptions.

`onSessionEvict`

Called when a session is evicted from an in-memory cache. This happens in two scenarios:

LRU eviction — a new session pushes an old one out of the cache (both the stateless API session registry and the UI agent cache).
Explicit removal — a user deletes a conversation.

Plugins that hold session-scoped resources (connection pools, open file handles, background tasks) should subscribe to this hook to clean them up.

Payload:

{"session_id": "abc123"}

Example handler:

async def register(self, bus):
    bus.on("onSessionEvict", self._on_session_evict, priority=10)

async def _on_session_evict(self, agent, ctx, payload):
    session_id = payload.get("session_id")
    if session_id:
        await self.connection_pool.close_session(session_id)
    return None  # fire-and-forget, no payload mutation needed

Note: agent is None for this hook since eviction is not tied to a specific agent turn. The ctx dict is empty. Only payload["session_id"] is meaningful.

Input & Context Hooks

Hook	Payload	Purpose
`onInputReceived`	`{raw: str}`	Intercept/modify raw user input
`onContextReady`	`{session_id, user_id, messages}`	Access loaded conversation context
`onPromptAssemble`	`{messages, tools}`	Modify prompt before LLM call

Prompt-patch lifetime: ephemeral, per turn

Terminology. A turn here is one user message plus the entire agent
loop it triggers, until the final response — the same sense the UI and
ctx["turn_id"] use. One LLM call within that loop is an iteration
(ctx["step"]): the iteration cap is max_iterations, the counter is
iterations_completed, and the per-iteration repeat guard is the
CrossIterationLoopDetector — all count iterations, not user rounds.
The legacy config key max_turns is still accepted as a deprecated alias
for max_iterations.

Anything a hook patches into messages is never persisted. The durable
transcript is written only at three explicit points — the user's input
(ConversationManager.prepare_conversation_context), the assistant's
response (store_assistant_response), and tool results
(ToolRuntime.store_results). Every new user turn rebuilds the prompt from
scratch: fresh system prompt from config, history loaded from storage, the
new user message. Hook patches decorate that in-flight list and vanish with
the turn — exactly like the system prompt itself.

Consequences:

A "skip if my marker is already in messages" check does not mean "inject once per conversation". It only dedupes across agent-loop iterations within one turn (the patched list is carried between iterations). Your patch will re-apply on every turn — design for that.
Re-injection per turn is the desired behavior for static context: config changes take effect on the next turn, and the stored transcript stays clean for export, sharing, and summarization.

Injecting context notes

add_context_note (from codumentor.agent.context_notes) is the single
primitive for putting standing or dynamic context in front of the model. It
owns placement, the synthetic_note tag, dedupe, and the caching/hijack
rules, so callers don't re-derive them:

from codumentor.agent.context_notes import add_context_note, content_freshness

async def _on_prompt_assemble(self, agent, ctx, payload):
    messages = payload.get("messages", [])
    note = add_context_note(
        messages,
        marker="[MyPlugin]",
        render=lambda: note_text,   # built only if the note is actually injected
        scope="prefix",             # "prefix" (static) | "pre_query" (dynamic)
        when=lambda a, c: bool(...),  # optional relevance gate; default always
    )
    if note is not None:            # None = gated out / already present this turn
        return {"type": "continue", "patch": {"messages": note}}

Pick the scope by whether the note's bytes are stable for the whole
conversation:

scope="prefix" — inserted directly after the leading system message. Static content only. A stable prefix keeps the assembled prompt prefix-stable across turns so provider prompt caching keeps working; a per-turn-varying prefix note invalidates that cache for the entire conversation (the framework logs a warning if a prefix note's bytes change turn-to-turn, when ctx is passed).
scope="pre_query" — inserted immediately before the current user query (the last non-synthetic user message). Use for anything that varies per turn or per loop iteration. It leaves the prefix byte-stable, and — because the real user query (and any tool work that follows) always comes after the note — the model never mistakes the note for the message it must answer. Do not append dynamic notes at the absolute end of the list: the moment a turn's real work finishes, a trailing user-role note becomes the dangling last message and the model "answers" it (this is exactly what bit repo_inventory — the agent acknowledged the repo list instead of confirming the completed task). The synthetic_note tag only protects internal last-user-message extractors (agentic-memory RDC retrieval skips tagged messages); the LLM never sees the tag, so position is the real safeguard. To let a note refresh during a turn when its content changes, pass freshness=content_freshness(note_text): the dedupe key becomes marker#freshness, so it re-injects when the content changes and is suppressed when it doesn't. (scope="tail" is a deprecated alias for "pre_query".)

Both scopes create a user-role message (models like Qwen on vLLM reject
non-leading system messages) tagged metadata={"synthetic_note": True}. A
note is injected at most once per marker within a turn — patches are
ephemeral and re-applied every turn (see "Prompt-patch lifetime"), so the
marker dedupes across loop iterations, not user turns.

All note callers — core and plugin, static and dynamic — go through
add_context_note directly; there is no longer a separate insert_context_note
shim (retired in the prompt-assembly redesign, migration step 7). For a static
note use scope="prefix"; for per-turn/per-iteration content use
scope="pre_query".

LLM Hooks

Hook	Payload	Purpose
`onLLMCallStart`	`{model, messages, tools, params}`	Before LLM API call
`onLLMStream`	`{stream_id, delta_text, ...}`	Real-time streaming control
`onLLMCallEnd`	`{content_chunks, tool_calls}`	After LLM response received

Tool Hooks

Hook	Payload	Purpose
`onToolInvokeStart`	`{tool, args, session_id}`	Before tool execution (includes permission wait)
`onToolExecuting`	`{tool, tool_call_id}`	When actual execution begins (after permission granted)
`onToolInvokeEnd`	`{tool, args, result, ms}`	After tool execution

Note: For tools requiring user permission (e.g., shell commands), there is a gap between onToolInvokeStart and onToolExecuting while waiting for user approval. Use onToolExecuting to measure actual execution time separately from wait time.

Response Hooks

Hook	Payload	Purpose
`onResponsePersist`	`{assistant_text, final_message?}`	After response is ready
`onBeforeSendResponse`	`{assistant_text, tool_calls, segments?}`	Final check / revise the finished answer
`onUsageTracked`	`{iteration_usage, session_totals}`	Token usage telemetry
`onReportCollect`	`{report_type, ctx_snapshot}`	Export data to reports
`onSessionReport`	`{format, agent_summary}`	Contribute sections to end-of-session report
`onSessionReportRender`	`{format, sections, agent_summary}`	Render collected session report sections

Revising the finished answer (`onBeforeSendResponse`)

A plugin can tune the assistant's finished answer — translate leaked text,
inject permalinks, redact secrets. Return a replace whose value.assistant_text
is the corrected whole text:

async def on_before_send(self, agent, ctx, payload):
    new_text = transform(payload["assistant_text"])
    return {"type": "replace", "value": {"assistant_text": new_text}}

This is persisted and replayed as a first-class revision (the canonical UI event
is assistant.message.revision), so the change survives reload — you do not
emit a stream chunk yourself. The whole machinery is described in
doc/core-dev/design/response-revision-persistence.md.

Multi-segment answers. When the turn streamed text → tool → text, the answer
is stored per segment, and a single whole-text blob can't be mapped back onto
those segments (earlier ones keep their old text, the last is sliced at the wrong
offset). For those turns the payload carries segments — the in-order non-empty
content segments. Return a per-segment corrected list alongside the whole
text so each persisted segment is revised in place:

segments = payload.get("segments")          # present only for multi-segment turns
if segments:                                # len >= 2, "".join(segments) == assistant_text
    value["segments"] = [transform(s) for s in segments]
return {"type": "replace", "value": value}

The agent diffs the list against the originals and emits one revision per changed
segment. A whole-text-only replace (no segments) keeps the legacy behavior and
is correct for single-segment answers.

Deferring the revision (optional). If the transformation is slow (e.g. an LLM
call), doing it inside onBeforeSendResponse blocks the turn tail. Instead you can
hand it to a background job so the turn finishes immediately and the revision lands
post-turn (placeholders briefly visible, then replaced live via SSE; reload is
correct too). Record a request on ctx["deferred_revision_jobs"] and return None:

ctx.setdefault("deferred_revision_jobs", []).append({
    "job_type": "my_translation_job",      # register the handler on the job queue (cf. agentic_memory)
    "queue": "background_llm",
    "args": {"conversation_id": ctx["conversation_id"],
             "message_id": ctx["message_id"], "turn_id": ctx.get("turn_id"), ...},
    "metadata": {"run_id": ctx.get("run_id")},
})
return None

After store_assistant_response, the agent enriches each request with the turn's
persisted-history descriptors and enqueues it; your handler does the work and calls
ctx.conv_service.revise_message(...) (the same primitive, segment-addressable) plus,
if needed, an in-place history rewrite via ConversationStorage.update_message_content
(pass expected_content to keep the rewrite drift-safe and idempotent on job re-runs).
script_filter (plugins/script_filter/translation_job.py, gated by its
defer_translation arg) is the reference implementation.

@-Mention Hooks

Hook	Payload	Purpose
`onAtQuery`	`{query, conversation_id}`	Provide dynamic @-mention search results
`onAtContentResolve`	`{references: [AtMentionItem, ...]}`	Resolve content for @-mentioned items

Note: These hooks use a collector pattern — multiple plugins can contribute results that are aggregated. The query and conversation_id are passed via the ctx dict rather than payload.

`onAtQuery`

Called when the user types @ in the chat input and the frontend requests dynamic results from plugins (via /ui/at-mentions/search/dynamic). Each plugin can return matching items for the query.

Return type:

return {"type": "at_items", "items": [
    AtMentionItem(
        id="unique_id",
        label="Display name",
        path="folder/file.py",
        type="file",           # Item type (e.g., "file", "single_file")
        provider="my_plugin",  # Your plugin identifier
        score=0.85,            # Relevance score for sorting (higher = better)
    ),
]}

Items from all plugins are collected, sorted by score, and returned to the frontend.

`onAtContentResolve`

Called when the agent needs to resolve the actual content of @-mentioned items before sending them to the LLM. The plugin should read/fetch the content for references matching its provider field.

Return type:

return {"type": "at_content", "resolved": {
    "item_id_1": "The full text content of the item...",
    "item_id_2": "Another item's content...",
}}

Resolved content from all plugins is merged into a single dict keyed by item ID.

Error & Control Hooks

Hook	Payload	Purpose
`onError`	`{error, context}`	Handle uncaught exceptions
`onAfterInput`	`{messages}`	Control flow after input
`onAfterLLM`	`{content_chunks, tool_calls, messages}`	Control flow after LLM
`onAfterTool`	`{tool, args, result, ms, messages}`	Control flow after tool
`onErrorDecision`	`{where, error, messages, last_action}`	Deterministic error recovery

Hook Return Values

Handlers return a dict to control the pipeline.

Basic Control

# Continue with no changes
return {"type": "continue"}

# Patch the payload (shallow merge)
return {"type": "continue", "patch": {"raw": "modified"}}

# Replace entire payload
return {"type": "replace", "value": new_payload}

# Return None for no-op
return None

Advanced Control (specific hooks only)

# Short-circuit the agent response (control hooks)
return {"type": "short_circuit", "assistant_text": "Early response"}

# Transform response (onBeforeSendResponse)
return {"type": "replace", "value": {"assistant_text": "Modified response"}}

# Retry tool with modified args (onToolInvokeEnd)
return {"type": "retry_with", "patched_args": {...}, "backoff_ms": 1000}

# Dynamic model routing (onLLMCallStart)
return {"type": "param_patch", "model": "gpt-4", "params": {"temperature": 0.2}}

# Stream control (onLLMStream)
return {"type": "stream_control", "abort": True, "replace_delta": "[FILTERED]"}

Return Type Summary

Return Type	Applicable Hooks	Effect
`continue`	All hooks	Continue pipeline, optionally with `patch`
`replace`	All hooks	Replace entire payload
`short_circuit`	`onAfterInput`, `onAfterLLM`, `onAfterTool`	End agent loop with given response
`retry_with`	`onToolInvokeEnd`	Retry the tool with patched arguments
`param_patch`	`onLLMCallStart`	Modify model or parameters before the call
`stream_control`	`onLLMStream`	Abort or replace streaming content
`at_items`	`onAtQuery`	Contribute @-mention search results (collector)
`at_content`	`onAtContentResolve`	Contribute resolved @-mention content (collector)

Emitting UI Events

To display something in the UI, emit events via the ui_event_emitter:

async def _on_response_persist(self, agent, ctx, payload):
    ui_event_emitter = ctx.get("ui_event_emitter")
    if ui_event_emitter:
        await ui_event_emitter(
            "plugin.my_plugin.completed",  # Event type
            {                               # Event data
                "status": "success",
                "summary": "Operation completed",
                "details": {...},
            },
            None  # Optional status override
        )

The event type should follow the convention plugin.<plugin_name>.<event_name>. The UI will route the event to your custom element based on the event_types declared in your UI manifest.

UI Manifest

Return UI configuration from the ui_manifest() method:

def ui_manifest(self) -> dict:
    return {
        "plugin_id": "my_plugin",
        "name": "My Plugin",
        "version": "1.0.0",
        "event_types": [
            "plugin.my_plugin.started",
            "plugin.my_plugin.completed",
        ],
        "ui": {
            "bundle_url": "/ui/extensions/plugins/my_plugin/index.js",
            "sdk_version": "1.0",
            "custom_elements": ["x-my-plugin-card"],
            "placements": ["transcript-card"],
            "widgets": [...]  # Optional welcome page widgets
        },
    }

Field	Required	Description
`plugin_id`	Yes	Internal identifier matching `meta["name"]`
`name`	No	Human-readable display name shown on the settings page. Falls back to `meta["name"]` if not provided.
`version`	No	Plugin version string
`event_types`	No	List of event types this plugin emits
`ui`	No	Web UI bundle configuration (see below)
`tui`	No	TUI (terminal) widget configuration (see below)
`settings_schema`	No	JSON Schema for user-configurable settings
`secret_requirements`	No	List of per-user secrets this plugin needs (see below). Surfaced by the User Secrets pane; does not create a settings panel.
`user_gates`	No	Array of per-user blocking-modal templates (EULA, privacy updates, …). See User Gates.

Declaring required secrets

If your plugin resolves a ${secret:<name>} placeholder (an API key, an auth
header, …), declare it under secret_requirements so the User Secrets
settings pane can show the user exactly which secret to create — and under what
name — instead of making them guess. This is descriptive metadata only: it is
read when the settings pane renders, never during a turn, so it adds **no
dependency on the user_secrets plugin and no load-order concern**.

from codumentor.core.secrets import SecretRequirement

def ui_manifest(self) -> dict:
    return {
        "plugin_id": "my_plugin",
        "name": "My Plugin",
        "secret_requirements": [
            SecretRequirement(
                name=self._api_key_secret,          # use the *configured* name
                label="My Service API key",
                description="Where the user finds it / what it's for.",
                required=True,
            ).as_dict(),
        ],
    }

`SecretRequirement` field	Description
`name`	The secret name as referenced via `${secret:<name>}`. Use the instance's configured name, not a hard-coded literal, so a renamed secret is reported truthfully.
`label`	Human-friendly label for the pane (falls back to `name`).
`description`	Optional help text — where to obtain the value, what it authorizes.
`required`	Whether the plugin is unusable without it (informational; blocks nothing).

Returning only plugin_id/name/secret_requirements (no settings_widget
or settings_schema) keeps the plugin out of the plugin-settings list — the
requirement is surfaced solely through the User Secrets pane. The host
aggregates declarations across all plugins via
PluginHost.collect_secret_requirements() (deduped by secret name), and the
user_secrets plugin annotates each with whether the user has set it. See the
User Secrets catalog entry.

Hint vs. resolution are separate concerns. Declaring a requirement only
makes the secret discoverable. Actually resolving ${secret:...} at
runtime still requires the user_secrets plugin to be loaded — it is what
publishes the per-turn registry. A plugin that resolves secrets during a turn
(after onContextReady) needs user_secrets loaded, but does not need a
dependencies: [user_secrets] edge purely to surface a requirement.

Plugins can optionally provide custom terminal UI widgets by declaring a "tui" section in their manifest. The TUI widget registry dynamically imports these at startup and uses them instead of the generic fallback card.

"tui": {
    "sdk_version": "1.0",
    "widget_module": "my_plugin.tui_widgets",    # Python import path
    "widget_classes": ["MyPluginCard"],           # Widget class names
    "placements": ["transcript-card"],            # Where to mount
}

Field	Required	Description
`sdk_version`	No	TUI SDK version (default: `"1.0"`)
`widget_module`	Yes	Dotted Python import path to the module containing widget classes
`widget_classes`	Yes	List of class names to import from the module
`placements`	No	Where widgets can be mounted: `"transcript-card"`, `"message-actions"`, `"status-bar"`, `"welcome"`, `"welcome-repos"`

Widget classes should subclass PluginCard (from codumentor.plugins.tui_sdk) for collapsible transcript cards, or BaseTUIPluginWidget (from codumentor.plugins.tui_sdk) for fully custom widgets.

For external plugins, the backend/ directory is automatically added to sys.path, so widget_module uses the same dotted path as any other module in the package.

If no "tui" section is declared, plugin events still render in the TUI using a generic fallback card that shows the plugin ID and data payload.

Agent-Specific Tool Registration

Control which agents receive your tools:

from codumentor.plugins.utils import should_register_tools_for_agent

class MyToolPlugin:
    def __init__(self, config=None, **kwargs):
        self.target_agents = kwargs.get('target_agents', 'all')

    async def _on_prompt_assemble(self, agent, ctx, payload):
        if not should_register_tools_for_agent(agent, self.target_agents):
            return None  # Skip for this agent type

        # Register tools...

Options:

"main": Only main agent
"subagent": Only subagents
"all": Both (default)