operative

operative

OperativeAgent — persistent, scheduled agent for autonomous operation.

Extends ToolUsingAgent with built-in session persistence and state recall. Designed for Operators: autonomous agents that run on a schedule with automatic state management between ticks.

Classes

OperativeAgent

OperativeAgent(engine: InferenceEngine, model: str, *, tools: Optional[List[BaseTool]] = None, bus: Optional[EventBus] = None, max_turns: int = 20, temperature: float = 0.3, max_tokens: int = 2048, system_prompt: Optional[str] = None, operator_id: Optional[str] = None, session_store: Optional[Any] = None, memory_backend: Optional[Any] = None, **kwargs: Any)

Bases: ToolUsingAgent

Persistent autonomous agent with built-in state management.

The Operative agent extends the standard tool-calling loop with:

1. Session loading — restores conversation history from previous ticks.
2. State recall — retrieves previous state JSON from memory backend.
3. System prompt — injects the operator's protocol instructions.
4. Tool loop — standard function-calling loop (same as Orchestrator).
5. Session save — persists the tick's prompt and response.
6. State persistence — auto-persists state if the agent didn't do it explicitly via memory_store tool.

Source code in src/openjarvis/agents/operative.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    tools: Optional[List[BaseTool]] = None,
    bus: Optional[EventBus] = None,
    max_turns: int = 20,
    temperature: float = 0.3,
    max_tokens: int = 2048,
    system_prompt: Optional[str] = None,
    operator_id: Optional[str] = None,
    session_store: Optional[Any] = None,
    memory_backend: Optional[Any] = None,
    **kwargs: Any,
) -> None:
    super().__init__(
        engine, model, tools=tools, bus=bus,
        max_turns=max_turns, temperature=temperature,
        max_tokens=max_tokens,
    )
    self._system_prompt = system_prompt or ""
    self._operator_id = operator_id
    self._session_store = session_store
    self._memory_backend = memory_backend

Functions

run

run(input: str, context: Optional[AgentContext] = None, **kwargs: Any) -> AgentResult

Execute a single operator tick.

Source code in src/openjarvis/agents/operative.py

def run(
    self,
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult:
    """Execute a single operator tick."""
    self._emit_turn_start(input)

    # 1. Build system prompt with state context
    sys_parts: list[str] = []
    if self._system_prompt:
        sys_parts.append(self._system_prompt)

    # 2. State recall from memory backend
    previous_state = self._recall_state()
    if previous_state:
        sys_parts.append(f"\n## Previous State\n{previous_state}")

    system_prompt = "\n\n".join(sys_parts) if sys_parts else None

    # 3. Load session history
    session_messages = self._load_session()

    # 4. Build messages
    messages = self._build_operative_messages(
        input, context, system_prompt=system_prompt,
        session_messages=session_messages,
    )

    # 5. Run function-calling tool loop
    openai_tools = self._executor.get_openai_tools() if self._tools else []
    all_tool_results: list[ToolResult] = []
    turns = 0
    content = ""
    state_stored_by_tool = False

    for _turn in range(self._max_turns):
        turns += 1

        if self._loop_guard:
            messages = self._loop_guard.compress_context(messages)

        gen_kwargs: dict[str, Any] = {}
        if openai_tools:
            gen_kwargs["tools"] = openai_tools

        result = self._generate(messages, **gen_kwargs)
        content = result.get("content", "")
        raw_tool_calls = result.get("tool_calls", [])

        if not raw_tool_calls:
            content = self._check_continuation(result, messages)
            break

        tool_calls = [
            ToolCall(
                id=tc.get("id", f"call_{i}"),
                name=tc.get("name", ""),
                arguments=tc.get("arguments", "{}"),
            )
            for i, tc in enumerate(raw_tool_calls)
        ]

        messages.append(Message(
            role=Role.ASSISTANT,
            content=content,
            tool_calls=tool_calls,
        ))

        for tc in tool_calls:
            # Loop guard check
            if self._loop_guard:
                verdict = self._loop_guard.check_call(tc.name, tc.arguments)
                if verdict.blocked:
                    tool_result = ToolResult(
                        tool_name=tc.name,
                        content=f"Loop guard: {verdict.reason}",
                        success=False,
                    )
                    all_tool_results.append(tool_result)
                    messages.append(Message(
                        role=Role.TOOL,
                        content=tool_result.content,
                        tool_call_id=tc.id,
                        name=tc.name,
                    ))
                    continue

            tool_result = self._executor.execute(tc)
            all_tool_results.append(tool_result)

            # Track if agent stored state via memory_store
            if tc.name == "memory_store" and self._operator_id:
                try:
                    args = json.loads(tc.arguments)
                    state_key = f"operator:{self._operator_id}:state"
                    if args.get("key", "") == state_key:
                        state_stored_by_tool = True
                except (json.JSONDecodeError, TypeError):
                    pass

            messages.append(Message(
                role=Role.TOOL,
                content=tool_result.content,
                tool_call_id=tc.id,
                name=tc.name,
            ))
    else:
        # Max turns exceeded
        self._save_session(input, content)
        return self._max_turns_result(all_tool_results, turns, content=content)

    # 6. Save session
    self._save_session(input, content)

    # 7. Auto-persist state if agent didn't do it explicitly
    if not state_stored_by_tool:
        self._auto_persist_state(content)

    self._emit_turn_end(turns=turns, content_length=len(content))
    return AgentResult(
        content=content,
        tool_results=all_tool_results,
        turns=turns,
    )