# With pip
pip install graphcrew

# With uv
uv add graphcrew

# OpenTelemetry instrumentation
pip install graphcrew[otel]

agents:
  - name: supervisor
    type: supervisor
    description: Routes user requests to the appropriate agent
  - name: greeter
    type: custom
    description: Greets users and tracks greeting count

topology:
  edges:
    - from_agent: supervisor
      to_agent: greeter
    - from_agent: greeter
      to_agent: supervisor
  default_return: supervisor

context:
  shared_slots:
    max_tokens: 4096
  eviction_strategy: lru

session:
  enabled: false

from graphcrew import AgentResponse, BaseAgent, OrchestratorState
from langchain_core.runnables import RunnableConfig


class GreeterAgent(BaseAgent):
    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        count = state.get("context_slots", {}).get("greeting_count", 0) + 1
        last_msg = state["messages"][-1].content if state["messages"] else ""

        content = f"Hello! This is greeting #{count}. You said: {last_msg}"

        return AgentResponse(
            agent_name=self.name,
            content=content,
            next_agent="supervisor",
            slots_update={"greeting_count": count},
        )

import asyncio
from langchain_openai import ChatOpenAI

from graphcrew import (
    AgentConfig,
    SupervisorAgent,
    TopologyResolver,
    create_initial_state,
    load_config,
)
from graphcrew.routing.graph_builder import build_hub_spoke_graph
from agent import GreeterAgent


async def main():
    config = load_config("config.yaml")
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    topology = TopologyResolver(config=config.topology)
    agent_configs = {a.name: a for a in config.agents}

    supervisor = SupervisorAgent(
        config=agent_configs["supervisor"],
        llm=llm,
        topology=topology,
    )
    greeter = GreeterAgent(config=agent_configs["greeter"])

    graph = build_hub_spoke_graph(supervisor, {"greeter": greeter}, topology)

    state = create_initial_state("Hello!", session_id="s1", user_id="alice")
    result = await graph.ainvoke(state)
    print(result["messages"][-1].content)


asyncio.run(main())

class BaseAgent(ABC):
    def __init__(
        self,
        config: AgentConfig,
        template_manager: ReactTemplateManager | None = None,
        react_executor: ReactExecutor | None = None,
        middleware: Sequence[AgentMiddleware] | None = None,
        *,
        redact_pii: bool = False,
    ) -> None: ...

    @abstractmethod
    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse: ...

    async def invoke(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> dict[str, Any]: ...

@dataclass(frozen=True, slots=True)
class AgentResponse:
    agent_name: str
    content: str
    next_agent: str | None = None           # None = terminate (no handoff)
    slots_update: dict[str, Any] = field(default_factory=dict)
    slots_tier: dict[str, SlotTier] = field(default_factory=dict)
    token_usage: dict[str, int] | None = None

class OrchestratorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    context_slots: Annotated[dict[str, Any], _merge_dicts]
    current_agent: str
    knowledge_context: Annotated[dict[str, Any], _merge_dicts]
    session_id: NotRequired[str]
    user_id: NotRequired[str]
    request_id: NotRequired[str]

state = create_initial_state(
    "Hello!",
    session_id="s1",         # default: "default"
    user_id="alice",          # default: "anonymous"
    current_agent="supervisor",  # default: "supervisor"
)

topology = TopologyResolver(config=config.topology)

topology.is_allowed("supervisor", "greeter")    # True if edge exists
topology.get_allowed_targets("supervisor")       # Sorted list, always includes default_return
topology.validate(["supervisor", "greeter"])     # Returns warnings for unreachable agents

topology:
  edges:
    - from_agent: supervisor
      to_agent: worker_a
    - from_agent: supervisor
      to_agent: worker_b
    - from_agent: worker_a
      to_agent: supervisor
    - from_agent: worker_b
      to_agent: supervisor
  default_return: supervisor

from graphcrew import load_config, load_config_from_string, load_config_from_dict

# From a file (accepts str or Path)
config = load_config("config.yaml")

# From a YAML string
config = load_config_from_string(yaml_str)

# From a dict
config = load_config_from_dict({"agents": [...], "topology": {...}})

agents:
  - name: supervisor
    type: supervisor
    description: Routes requests to the right agent
    model: gpt-4o           # optional LLM model override
    knowledge_enabled: true  # enable knowledge module system
  - name: researcher
    type: custom
    description: Performs research tasks

topology:
  edges:
    - from_agent: supervisor
      to_agent: researcher
    - from_agent: researcher
      to_agent: supervisor
  default_return: supervisor

context:
  shared_slots:
    max_tokens: 4096        # token budget for the slot pool (must be > 0)
  tiered_slots:
    turn:
      max_tokens: 4096
    session:
      max_tokens: 4096
    long_term:
      max_tokens: 4096
  eviction_strategy: lru    # lru | fifo | priority | demotion

session:
  enabled: true

knowledge:
  enabled: true
  base_dir: llm_resources   # root directory for agent knowledge folders
  max_total_tokens: 8192    # token budget for loaded knowledge (must be > 0)

retry:
  initial_interval: 0.5     # seconds before first retry (must be > 0)
  backoff_factor: 2.0       # exponential base multiplier
  max_interval: 128.0       # maximum seconds between retries
  max_attempts: 3           # total attempts including first (1 = no retry)
  jitter: true              # random jitter to prevent thundering herd

timeout:
  llm_call_seconds: 30.0    # per-call LLM timeout
  file_io_seconds: 5.0      # per-file knowledge load timeout
  persistence_seconds: 10.0 # per-call slot persistence timeout
  total_request_seconds: 120.0 # total timeout across all ReAct passes

from graphcrew import AgentConfig, AgentResponse, BaseAgent, OrchestratorState, SlotTier
from langchain_core.runnables import RunnableConfig


class SummarizerAgent(BaseAgent):
    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        messages = state["messages"]
        # Your summarization logic here
        summary = f"Summarized {len(messages)} messages."

        return AgentResponse(
            agent_name=self.name,
            content=summary,
            next_agent=None,  # None = terminate, graph ends
            slots_update={"last_summary": summary},
            slots_tier={"last_summary": SlotTier.SESSION},
        )


# Instantiate with a config
agent = SummarizerAgent(
    config=AgentConfig(name="summarizer", type="custom", description="Summarizes conversations")
)

from graphcrew import (
    SupervisorAgent,
    AgentConfig,
    TopologyResolver,
    RetryConfig,
    TimeoutConfig,
    SemaphoreRateLimiter,
    MetricsMiddleware,
    InMemoryMetricsCollector,
)

collector = InMemoryMetricsCollector()

supervisor = SupervisorAgent(
    config=AgentConfig(name="supervisor", type="supervisor", description="Routes requests"),
    llm=llm,
    topology=topology,
    retry_config=RetryConfig(max_attempts=3),
    timeout_config=TimeoutConfig(llm_call_seconds=30.0),
    rate_limiter=SemaphoreRateLimiter(max_concurrency=5),
    use_structured_output=True,
    middleware=[MetricsMiddleware(collector)],
    redact_pii=True,
)

from graphcrew.routing.graph_builder import build_hub_spoke_graph

graph = build_hub_spoke_graph(
    supervisor=supervisor,
    agents={"greeter": greeter, "researcher": researcher},
    topology=topology,
)
result = await graph.ainvoke(create_initial_state("Hello"))

topology:
  edges:
    - from_agent: supervisor
      to_agent: greeter
    - from_agent: supervisor
      to_agent: researcher
    - from_agent: greeter
      to_agent: supervisor
    - from_agent: researcher
      to_agent: supervisor
  default_return: supervisor

from graphcrew.routing.graph_builder import build_topology_graph

graph = build_topology_graph(
    agents={
        "supervisor": supervisor,
        "billing": billing,
        "support": support,
    },
    topology=topology,
    entry_point="supervisor",
)

topology:
  edges:
    - from_agent: supervisor
      to_agent: billing
    - from_agent: supervisor
      to_agent: support
    - from_agent: billing
      to_agent: support     # direct worker-to-worker
    - from_agent: support
      to_agent: supervisor
    - from_agent: billing
      to_agent: supervisor
  default_return: supervisor

from graphcrew.routing.graph_builder import build_conditional_edge

routing_fn = build_conditional_edge(topology, "supervisor")
# Returns END when state["current_agent"] == "supervisor" (termination sentinel)
# Falls back to topology.default_return for unknown targets

from graphcrew import SlotManager, SlotTier, ContextConfig

manager = SlotManager(config=ContextConfig())

# Shared slots (default tier: TURN)
manager.set_shared("user_name", "Alice")
manager.set_shared_tiered("preference", "dark_mode", tier=SlotTier.SESSION)

# Per-agent slots
manager.set_agent("researcher", "search_query", "quantum computing")

# Promote a slot to a higher tier
manager.promote_slot("user_name", SlotTier.LONG_TERM)

# Serialization
data = manager.to_dict()     # {"shared": {...}, "agents": {...}}
manager.from_dict(data)       # Restore from dict

# Lifecycle
async with SlotManager(config=ContextConfig()) as mgr:
    mgr.set_shared("key", "value")

return AgentResponse(
    agent_name=self.name,
    content="Done",
    slots_update={"result": computed_value},
    slots_tier={"result": SlotTier.SESSION},
)

context:
  eviction_strategy: demotion  # or lru, fifo, priority

from graphcrew import EvictionStrategy, Slot

class CustomEviction(EvictionStrategy):
    def select(self, slots: list[Slot], count: int) -> list[Slot]:
        # Return up to `count` slots to evict
        return sorted(slots, key=lambda s: s.priority)[:count]

from graphcrew import SlotPersistence, Slot

class PostgresSlotPersistence(SlotPersistence):
    async def load_session_slots(self, session_id: str) -> list[Slot]: ...
    async def save_session_slots(self, session_id: str, slots: list[Slot]) -> None: ...
    async def delete_session_slots(self, session_id: str) -> None: ...
    async def load_long_term_slots(self, user_id: str) -> list[Slot]: ...
    async def save_long_term_slots(self, user_id: str, slots: list[Slot]) -> None: ...
    async def delete_long_term_slots(self, user_id: str) -> None: ...

from graphcrew import InMemorySlotPersistence, SlotManager, ContextConfig

persistence = InMemorySlotPersistence()
manager = SlotManager(config=ContextConfig(), persistence=persistence)

await manager.hydrate(session_id="s1", user_id="alice")
# ... run your graph ...
await manager.flush(session_id="s1", user_id="alice")
await manager.clear_turn_slots()

llm_resources/
  supervisor/
    react_config.yaml
    react_core.md
    slots/
      routing_rules.md
      escalation_policy.md

max_passes: 5

actions:
  request_slot:
    description: Load a knowledge slot for reference
    parameters:
      slot: Name of the slot to load
    required:
      - slot
  decide:
    description: Make a final routing decision
    parameters:
      next_agent: The agent to route to
      content: Summary of reasoning
    required:
      - next_agent

slots:
  routing_rules:
    description: Rules for routing requests to agents
    modules:
      - slots/routing_rules.md
    triggers: When deciding which agent should handle the request
    max_tokens: 2000
  escalation_policy:
    description: Escalation rules
    modules:
      - slots/escalation_policy.md
    dependencies:
      - routing_rules

from graphcrew import (
    ReactExecutor,
    ReactTemplateManager,
    FileSystemKnowledgeLoader,
    RetryConfig,
    TimeoutConfig,
)

loader = FileSystemKnowledgeLoader(base_dir="llm_resources")
template_manager = ReactTemplateManager(
    agent_name="supervisor",
    base_dir="llm_resources",
    loader=loader,
    max_loaded_tokens=8192,
)

executor = ReactExecutor(
    agent_name="supervisor",
    llm=llm,
    template_manager=template_manager,
    action_handlers={"request_slot": slot_handler, "decide": decide_handler},
    max_passes=5,
    retry_config=RetryConfig(max_attempts=3),
    timeout_config=TimeoutConfig(llm_call_seconds=30.0, total_request_seconds=120.0),
    max_history_tokens=2000,
    rate_limiter=limiter,
    use_structured_output=True,
)

result = await executor.execute(user_input="Hello", context={"key": "value"}, config=config)

class ActionHandler(Protocol):
    async def __call__(
        self,
        params: dict[str, Any],
        state: dict[str, Any],
        config: RunnableConfig | None = None,
    ) -> tuple[str, dict[str, Any]]: ...

from graphcrew import InMemorySessionManager, SessionMessage, Memory, Summary

session_mgr = InMemorySessionManager()

# Save messages
await session_mgr.save_messages("s1", [
    SessionMessage(role="user", content="Hello", session_id="s1", user_id="alice"),
])

# Get messages (limit must be >= 1)
messages = await session_mgr.get_messages("s1", limit=50, user_id="alice")

# Store and recall memories
await session_mgr.store_memory("s1", Memory(
    session_id="s1", key="preference", value="dark mode",
    importance=0.8, user_id="alice",
), user_id="alice")

memory = await session_mgr.recall_memory("s1", "preference")

# Search memories (sorted by importance descending, expired memories excluded)
results = await session_mgr.search_memories("s1", "preference", limit=5)

# Cross-session search by user
results = await session_mgr.search_memories_by_user("alice", "preference", limit=10)

# Pagination
page = await session_mgr.get_messages_paginated("s1", cursor=None, limit=20)
# page.items, page.next_cursor, page.has_more

# Update memory (only content fields: value, importance, source_agent, categories, metadata, expires_at)
updated = await session_mgr.update_memory("s1", memory.id, {"importance": 0.9}, user_id="alice")

# Delete memory (with ownership check)
deleted = await session_mgr.delete_memory("s1", memory.id, user_id="alice")

# Lifecycle
await session_mgr.clear_session("s1")  # Clears data but retains session lock
await session_mgr.close()              # Sets _closed flag, clears all data

caps = session_mgr.get_capabilities()
# InMemorySessionManager returns all 13 methods

async with InMemorySessionManager() as mgr:
    await mgr.save_messages("s1", messages)

from graphcrew import TenantIsolationMiddleware

middleware = TenantIsolationMiddleware()

agent = MyAgent(
    config=agent_config,
    middleware=[middleware],
)

from graphcrew import AgentMiddleware, OrchestratorState, AgentResponse
from langchain_core.runnables import RunnableConfig


class LoggingMiddleware(AgentMiddleware):
    async def before_invoke(
        self,
        agent_name: str,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> OrchestratorState:
        print(f"Before: {agent_name}")
        return state

    async def after_invoke(
        self,
        agent_name: str,
        state: OrchestratorState,
        response: AgentResponse,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        print(f"After: {agent_name} -> {response.next_agent}")
        return response

    async def on_error(
        self,
        agent_name: str,
        state: OrchestratorState,
        error: Exception,
        config: RunnableConfig | None = None,
    ) -> None:
        print(f"Error: {agent_name}: {error}")

from graphcrew import MetricsMiddleware, InMemoryMetricsCollector

collector = InMemoryMetricsCollector()
metrics_mw = MetricsMiddleware(collector)

agent = MyAgent(config=agent_config, middleware=[metrics_mw])

class MetricsCollector(Protocol):
    def record_invocation(self, record: InvocationRecord) -> None: ...
    def record_error(self, record: ErrorRecord) -> None: ...

from graphcrew import CircuitBreakerMiddleware

cb = CircuitBreakerMiddleware(
    failure_threshold=5,          # failures before opening
    recovery_timeout_seconds=60,  # seconds before half-open probe
)

agent = MyAgent(config=agent_config, middleware=[cb])

from graphcrew import MemoryInjectionMiddleware, InMemorySessionManager

session_mgr = InMemorySessionManager()
memory_mw = MemoryInjectionMiddleware(session_mgr, limit=5)

supervisor = SupervisorAgent(
    config=agent_config,
    llm=llm,
    topology=topology,
    middleware=[memory_mw],
)

from graphcrew import SemaphoreRateLimiter

limiter = SemaphoreRateLimiter(max_concurrency=5, delay_seconds=0.1)

supervisor = SupervisorAgent(
    config=agent_config, llm=llm, topology=topology,
    rate_limiter=limiter,
)

from graphcrew import PerTenantRateLimiter

limiter = PerTenantRateLimiter(
    max_concurrency=5,
    delay_seconds=0.0,
    max_tenants=10_000,
)

# In your request handler:
limiter.set_tenant("tenant-abc")
await limiter.acquire()
try:
    # ... LLM call ...
finally:
    await limiter.release()

async with SemaphoreRateLimiter(max_concurrency=5) as limiter:
    # ...

async for chunk_meta, accumulated in supervisor.astream_process(state, config):
    if chunk_meta["chunk_type"] == "token":
        print(chunk_meta["delta"], end="", flush=True)
    elif chunk_meta["chunk_type"] == "final":
        print(f"\nRouting to: {chunk_meta['next_agent']}")

from graphcrew import astream_with_timeout

async for chunk in astream_with_timeout(llm, messages, timeout_seconds=30.0, config=config):
    print(chunk.content, end="")

executor = ReactExecutor(
    agent_name="supervisor",
    llm=llm,
    template_manager=template_manager,
    action_handlers=handlers,
    use_structured_output=True,
)

class ReactStructuredResponse(BaseModel):
    thought: str
    action: str
    action_params: dict[str, Any] = Field(default_factory=dict)

pip install graphcrew[otel]

from graphcrew.observability import (
    OTEL_AVAILABLE,
    OpenTelemetryMiddleware,
    get_tracer,
    otel_span,
    sync_otel_span,
)

# Middleware: creates spans around agent invocations
otel_mw = OpenTelemetryMiddleware()
agent = MyAgent(config=agent_config, middleware=[otel_mw])

# Manual spans
async with otel_span("my_operation", attributes={"key": "value"}) as span:
    # ... your code ...
    pass

with sync_otel_span("react_pass", agent_name="supervisor", pass_num=1):
    # ... your code ...
    pass

from graphcrew import bind_logging_context, clear_logging_context

# bind_logging_context is called automatically by BaseAgent.invoke()
# It binds: request_id, session_id, user_id, langgraph_step, langgraph_node

# App developers must configure structlog with merge_contextvars:
import structlog
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.dev.ConsoleRenderer(),
    ],
)

# Clear at request boundaries:
clear_logging_context()

agent = MyAgent(config=agent_config, redact_pii=True)

from graphcrew import TokenCounter, CharDivisionTokenCounter

# Default: len(text) // 4, minimum 1 for non-empty text, 0 for empty
counter = CharDivisionTokenCounter(chars_per_token=4)
counter.count("hello world")  # 2

# Custom counter (e.g., tiktoken-based)
class TiktokenCounter(TokenCounter):
    def __init__(self, model: str = "gpt-4o"):
        import tiktoken
        self._enc = tiktoken.encoding_for_model(model)

    def count(self, text: str) -> int:
        if not text:
            return 0
        return len(self._enc.encode(text))

slot_pool = SlotPool(token_counter=counter)
slot_manager = SlotManager(config=context_config, token_counter=counter)
template_manager = ReactTemplateManager(
    agent_name="supervisor", base_dir="llm_resources",
    loader=loader, token_counter=counter,
)

OrchestratorError
  ConfigurationError
    ConfigLoadError
    ConfigValidationError
    LLMNotConfiguredError
    ReactExecutorNotConfiguredError
  AgentExecutionError
    AgentProcessError
  LLMError
    RateLimitExceededError
    LLMInvocationError
    LLMRetryExhaustedError
  SessionError
    SessionManagerError
    SessionManagerNotConfiguredError
  StateError
    InvalidStateError
  ContextError
    SlotPersistenceError
  KnowledgeError
    KnowledgeSlotNotFoundError
    KnowledgePathError
    KnowledgeLoadError
    KnowledgeConfigValidationError
  ReactError
    ReactCancellationError
    ReactMaxPassesError
    ReservedStateKeyError
    ReactConsecutiveErrorsError
  RoutingError
    RoutingSelfLoopError
  TenantIsolationError
  CircuitBreakerOpenError

# SessionManager
status = await session_mgr.health_check()
# InMemorySessionManager: {"status": "ok", "sessions": 3}
# When closed: {"status": "closed", "sessions": 0}

# SlotPersistence
status = await persistence.health_check()
# InMemorySlotPersistence: {"status": "ok", "session_slots": 5, "long_term_slots": 2}

from graphcrew import (
    register_transient_exception,
    deregister_transient_exception,
    get_transient_exceptions,
)

# At app startup
from anthropic import RateLimitError
register_transient_exception(RateLimitError)

# Check current set
print(get_transient_exceptions())  # (OSError, TimeoutError, RateLimitError)

# Deregister (safe in test teardown, no-op if not registered)
deregister_transient_exception(RateLimitError)

from graphcrew import SemanticSearchMixin, InMemorySessionManager, Memory

class MySemanticSessionManager(SemanticSearchMixin, InMemorySessionManager):
    async def embed_text(self, text: str) -> list[float]:
        # Your embedding provider
        return await my_embedder.embed(text)

    async def search_memories_semantic(
        self, session_id: str, query: str, limit: int = 5,
    ) -> list[Memory]:
        query_vec = await self.embed_text(query)
        # Your vector similarity search implementation
        return await self._vector_store.search(session_id, query_vec, limit)

from graphcrew import OrchestratorContext

ctx = OrchestratorContext(
    slot_manager=slot_mgr,
    session_manager=session_mgr,
    template_managers=[tm],
    rate_limiter=limiter,
    extras=[custom_resource],
)

async with ctx:
    # Run your graph
    result = await graph.ainvoke(state)

# ctx.close() tears down in safe order:
# extras -> template managers -> rate limiter -> slot manager -> session manager
# Errors are logged but never mask other component cleanup.

# Check status
print(ctx.is_closed)  # True after close()

from graphcrew.testing import MockLLM, make_test_state, InMemorySessionManager

# MockLLM returns scripted responses
llm = MockLLM(responses=[
    '<thought>Routing to greeter</thought><action type="decide">{"next_agent": "greeter"}</action>',
    "Hello!",
])

# Raises IndexError when exhausted
# Optional: raise_on_call=ValueError to test error handling
# Optional: stream_chunks=[["chunk1", "chunk2"]] for astream testing
# Optional: structured_responses=[obj] for with_structured_output testing
# Optional: usage_tokens={"input_tokens": 10, "output_tokens": 5}

# make_test_state creates a pre-populated OrchestratorState
state = make_test_state("Hello")
# Defaults: session_id="test-session", user_id="test-user"
state = make_test_state("Hello", session_id="s1", user_id="alice")

# InMemorySessionManager re-exported for convenience
session_mgr = InMemorySessionManager()

from graphcrew.routing.graph_builder import (
    build_hub_spoke_graph,
    build_topology_graph,
    build_conditional_edge,
)

# Hub-spoke: all workers route back to supervisor
graph = build_hub_spoke_graph(supervisor, {"greeter": greeter}, topology)

# Topology-driven: edges match topology exactly (supports mesh)
graph = build_topology_graph(
    agents={"supervisor": supervisor, "billing": billing, "support": support},
    topology=topology,
    entry_point="supervisor",
)

# Manual edge wiring
routing_fn = build_conditional_edge(topology, "supervisor")
builder.add_conditional_edges("supervisor", routing_fn, edge_map)

                  +-----------+
            +---->| monitor   |----+
            |     +-----------+    |
            |                      v
    +------------+           +--------------+
    | supervisor |<--------->| diagnostician|
    +------------+           +--------------+
            |                      |
            |     +-----------+    |
            +---->| remediator|<---+
            |     +-----------+
            |          |
            |     +-----------+
            +---->| reporter  |
                  +-----------+

# config.yaml -- DevOps Incident Response System

agents:
  - name: supervisor
    type: supervisor
    description: >-
      Orchestrates the incident response workflow. Routes incoming alerts to
      the monitor for classification, then to diagnostician for root-cause
      analysis, remediator for fixes, and reporter for post-mortems.
    knowledge_enabled: true

  - name: monitor
    type: custom
    description: >-
      Detects and classifies incidents from raw alert payloads. Determines
      severity (P1-P4) and affected service.

  - name: diagnostician
    type: custom
    description: >-
      Analyzes classified incidents to determine root cause. Uses runbook
      knowledge modules for system-specific diagnostic procedures.
    knowledge_enabled: true

  - name: remediator
    type: custom
    description: >-
      Executes remediation playbooks based on diagnosis. Uses playbook
      knowledge modules for step-by-step remediation procedures.
    knowledge_enabled: true

  - name: reporter
    type: custom
    description: >-
      Generates structured incident reports and post-mortem documents.

topology:
  edges:
    - from_agent: supervisor
      to_agent: monitor
    - from_agent: supervisor
      to_agent: diagnostician
    - from_agent: supervisor
      to_agent: remediator
    - from_agent: supervisor
      to_agent: reporter
    - from_agent: monitor
      to_agent: diagnostician
    - from_agent: monitor
      to_agent: supervisor
    - from_agent: diagnostician
      to_agent: remediator
    - from_agent: diagnostician
      to_agent: supervisor
    - from_agent: remediator
      to_agent: reporter
    - from_agent: remediator
      to_agent: supervisor
    - from_agent: reporter
      to_agent: supervisor
  default_return: supervisor

context:
  shared_slots:
    max_tokens: 8192
  tiered_slots:
    turn:
      max_tokens: 4096
    session:
      max_tokens: 8192
    long_term:
      max_tokens: 4096
  eviction_strategy: lru

session:
  enabled: true

knowledge:
  enabled: true
  base_dir: knowledge
  max_total_tokens: 16384

retry:
  initial_interval: 1.0
  backoff_factor: 2.0
  max_interval: 30.0
  max_attempts: 3
  jitter: true

timeout:
  llm_call_seconds: 30.0
  total_request_seconds: 120.0

"""Custom agents for the DevOps Incident Response System."""
from __future__ import annotations

import json
from typing import Any

from langchain_core.runnables import RunnableConfig

from graphcrew import AgentResponse, BaseAgent, OrchestratorState
from graphcrew.context.slots import SlotTier


class MonitorAgent(BaseAgent):
    """Classifies raw alerts into structured incidents."""

    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        messages = state.get("messages", [])
        alert_text = str(messages[-1].content) if messages else "unknown alert"

        severity = "P2"
        service = "unknown"
        category = "infrastructure"

        if "cpu" in alert_text.lower() or "memory" in alert_text.lower():
            category = "resource_exhaustion"
            service = "compute"
        elif "latency" in alert_text.lower() or "timeout" in alert_text.lower():
            category = "performance"
            service = "api-gateway"
        elif "error rate" in alert_text.lower() or "5xx" in alert_text.lower():
            category = "availability"
            service = "web-service"
            severity = "P1"

        classification = {
            "severity": severity,
            "service": service,
            "category": category,
            "raw_alert": alert_text[:500],
        }

        return AgentResponse(
            agent_name=self.name,
            content=f"Incident classified: {severity} {category} on {service}",
            next_agent="diagnostician",
            slots_update={
                "incident_classification": json.dumps(classification),
                "incident_severity": severity,
                "incident_service": service,
            },
            slots_tier={
                "incident_severity": SlotTier.SESSION,
                "incident_service": SlotTier.SESSION,
            },
        )


class DiagnosticianAgent(BaseAgent):
    """Performs root-cause analysis using context from the monitor."""

    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        slots = state.get("context_slots", {})
        classification_raw = slots.get("incident_classification", "{}")
        classification = json.loads(classification_raw) if isinstance(
            classification_raw, str
        ) else classification_raw

        category = classification.get("category", "unknown")
        service = classification.get("service", "unknown")

        root_cause = "unknown"
        recommended_action = "escalate"

        if category == "resource_exhaustion":
            root_cause = f"Resource limits exceeded on {service}"
            recommended_action = "scale_horizontally"
        elif category == "performance":
            root_cause = f"Latency degradation in {service}"
            recommended_action = "circuit_breaker_and_retry"
        elif category == "availability":
            root_cause = f"Service {service} returning errors due to deployment regression"
            recommended_action = "rollback_deployment"

        diagnosis = {
            "root_cause": root_cause,
            "recommended_action": recommended_action,
            "confidence": 0.85,
        }

        return AgentResponse(
            agent_name=self.name,
            content=f"Diagnosis: {root_cause}. Recommended: {recommended_action}",
            next_agent="remediator",
            slots_update={"incident_diagnosis": json.dumps(diagnosis)},
            slots_tier={"incident_diagnosis": SlotTier.SESSION},
        )


class RemediatorAgent(BaseAgent):
    """Executes remediation based on the diagnosis."""

    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        slots = state.get("context_slots", {})
        diagnosis_raw = slots.get("incident_diagnosis", "{}")
        diagnosis = json.loads(diagnosis_raw) if isinstance(
            diagnosis_raw, str
        ) else diagnosis_raw

        recommended_action = diagnosis.get("recommended_action", "escalate")
        status = "completed"
        steps = ["Executed remediation steps"]

        remediation = {
            "action_taken": recommended_action,
            "steps": steps,
            "status": status,
        }

        return AgentResponse(
            agent_name=self.name,
            content=f"Remediation {status}: {recommended_action}",
            next_agent="reporter",
            slots_update={"incident_remediation": json.dumps(remediation)},
            slots_tier={"incident_remediation": SlotTier.SESSION},
        )


class ReporterAgent(BaseAgent):
    """Generates incident reports from the full incident context."""

    async def process(
        self,
        state: OrchestratorState,
        config: RunnableConfig | None = None,
    ) -> AgentResponse:
        slots = state.get("context_slots", {})
        classification = json.loads(slots.get("incident_classification", "{}"))
        diagnosis = json.loads(slots.get("incident_diagnosis", "{}"))
        remediation = json.loads(slots.get("incident_remediation", "{}"))

        report = f"""# Incident Report
**Severity:** {classification.get('severity', 'P3')}
**Service:** {classification.get('service', 'unknown')}
## Root Cause
{diagnosis.get('root_cause', 'unknown')}
## Remediation
**Action:** {remediation.get('action_taken', 'none')}
**Status:** {remediation.get('status', 'unknown')}"""

        return AgentResponse(
            agent_name=self.name,
            content=report,
            next_agent=None,  # Terminates the graph
            slots_update={"incident_report": report},
            slots_tier={"incident_report": SlotTier.LONG_TERM},
        )

"""FastAPI application for the DevOps Incident Response System."""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from pathlib import Path

import structlog
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

from graphcrew import (
    AgentConfig,
    CircuitBreakerOpenError,
    InMemoryMetricsCollector,
    OrchestratorContext,
    OrchestratorError,
    PerTenantRateLimiter,
    RateLimitExceededError,
    SupervisorAgent,
    TopologyResolver,
    clear_logging_context,
    create_initial_state,
    load_config,
)
from graphcrew.routing.graph_builder import build_topology_graph

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer(),
    ],
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage OrchestratorContext lifecycle."""
    config = load_config(Path("config.yaml"))
    rate_limiter = PerTenantRateLimiter(max_concurrency=5, max_tenants=1000)

    ctx = OrchestratorContext(rate_limiter=rate_limiter)
    app.state.ctx = ctx
    yield
    await ctx.close()


app = FastAPI(
    title="DevOps Incident Response System",
    lifespan=lifespan,
)


@app.post("/incident")
async def create_incident(
    alert: str,
    x_user_id: str = Header(..., alias="X-User-Id"),
):
    clear_logging_context()
    session_id = f"incident-{uuid.uuid4()}"
    state = create_initial_state(alert, session_id=session_id, user_id=x_user_id)

    try:
        result = await app.state.graph.ainvoke(state)
        return {"incident_id": session_id, "report": result["messages"][-1].content}
    except RateLimitExceededError:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    except CircuitBreakerOpenError:
        raise HTTPException(status_code=503, detail="Service temporarily unavailable")
    except OrchestratorError as exc:
        raise HTTPException(status_code=500, detail=str(exc))


@app.get("/health")
async def health():
    ctx = app.state.ctx
    return {"status": "ok" if not ctx.is_closed else "closed"}

# Install dependencies
pip install graphcrew fastapi uvicorn langchain-openai

# Start the server
uvicorn devops_incident.app:app --host 0.0.0.0 --port 8000

# Submit an incident
curl -X POST http://localhost:8000/incident \
  -H "Content-Type: application/json" \
  -H "X-User-Id: ops-team" \
  -d '{"alert": "CRITICAL: 5xx error rate above 10% on web-service"}'

# Health check
curl http://localhost:8000/health

git clone https://github.com/Atiqul-Islam/swagent.git
cd swagent

# Create virtual environment and install all deps (including dev)
uv sync --all-extras --dev

# Run the full quality gate (lint + typecheck + tests)
make all

# Or run individually:
uv run ruff check src/ tests/          # Linting
uv run ruff format --check src/ tests/ # Formatting
uv run mypy src/ tests/                # Type checking (strict)
uv run pytest                          # 1300+ tests, 98%+ coverage

# OpenTelemetry instrumentation support
uv sync --extra otel

+---------------------------------------------------------------------------+
|                  graphcrew (library)                                |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  1. Configuration                                                    | |
|  |  YAML -> Pydantic -> agents, topology, slots, knowledge             | |
|  +----------------------------+----------------------------------------+ |
|                               |                                           |
|                               v                                           |
|  +---------------------------------------------------------------------+ |
|  |  2. Core                                                             | |
|  |  OrchestratorState, types, exceptions                                | |
|  +--------+------------------------------+-----------------------------+ |
|           |                              |                                |
|           v                              v                                |
|  +----------------+       +-------------------------------------------+  |
|  | 3. Built-in    |       | Routing                                   |  |
|  | Agents         |       | TopologyResolver (validates)              |  |
|  |                |       |                                           |  |
|  | Supervisor     |       |                                           |  |
|  | SessionMgr     |       +-------------------------------------------+  |
|  +----------------+                                                       |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  4. Slot-Based Context Management                                    | |
|  |  SlotManager -> per-agent pools + shared pool                        | |
|  |  SlotTier (TURN / SESSION / LONG_TERM)                               | |
|  |  TokenCounter (ABC) -> pluggable token estimation                    | |
|  |  EvictionStrategy (ABC) -> LRU / FIFO / Priority / Demotion         | |
|  |  SlotPersistence (ABC) -> pluggable durable storage                  | |
|  +---------------------------------------------------------------------+ |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  5. Session Layer                                                     | |
|  |  SessionManager (ABC: 3 abstract methods)                            | |
|  |  Summaries + memories default to NotImplementedError                 | |
|  |  InMemorySessionManager ships for dev/testing (per-session locks)    | |
|  +---------------------------------------------------------------------+ |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  6. Knowledge Module System                                          | |
|  |  KnowledgeLoader (ABC) + FileSystemKnowledgeLoader                   | |
|  |  ReactTemplateManager (per-agent config + slot loading)              | |
|  |  ReactConfigSchema (Pydantic validation for react_config.yaml)       | |
|  +---------------------------------------------------------------------+ |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  7. ReAct Execution Engine                                           | |
|  |  ReactExecutor (multi-pass loop + token-budgeted history window)    | |
|  |  ReactParsedResponse, TerminalActionResult, ActionHandler            | |
|  +---------------------------------------------------------------------+ |
|                                                                           |
|  +---------------------------------------------------------------------+ |
|  |  8. Extension Points (what developers implement)                     | |
|  |  BaseAgent (ABC) | SessionManager (ABC) | KnowledgeLoader (ABC)     | |
|  |  EvictionStrategy (ABC) | SlotPersistence (ABC) | TokenCounter (ABC)| |
|  |  AgentMiddleware | RateLimiter (ABC) | YAML config                | |
|  +---------------------------------------------------------------------+ |
+---------------------------------------------------------------------------+

User Input
    |
    v
[Supervisor Node] --- reads context slots ---> [SlotManager]
    |                                               |
    | (conditional edge: route_decision)            | (optional)
    v                                               v
[Agent-X Node] <--- historical context --- [SessionManagerAgent]
    |                                           |
    | (handoff / return)                        v
    v                                     [SessionManager impl]
[Supervisor Node]                          (developer-provided)
    |
    v
Final Response

class OrchestratorConfig(BaseModel):
    """Top-level config. frozen=True, extra='forbid'."""
    agents: list[AgentConfig]
    topology: TopologyConfig
    context: ContextConfig
    session: SessionConfig
    knowledge: KnowledgeConfig
    retry: RetryConfig
    timeout: TimeoutConfig

class AgentConfig(BaseModel):
    """Per-agent config with name validation."""
    name: str          # [a-zA-Z][a-zA-Z0-9_-]*, <= 64 chars, no reserved names
    type: Literal["supervisor", "session_manager", "custom"]
    description: str   # min_length=1
    model: str | None
    knowledge_enabled: bool

class TopologyConfig(BaseModel):
    edges: list[TopologyEdgeConfig]
    default_return: str  # default "supervisor"

class TimeoutConfig(BaseModel):
    llm_call_seconds: float | None
    file_io_seconds: float | None
    persistence_seconds: float | None
    total_request_seconds: float | None

class RetryConfig(BaseModel):
    initial_interval: float   # 0.5
    backoff_factor: float     # 2.0
    max_interval: float       # 128.0
    max_attempts: int         # 3 (ge=1; 1 = no retry)
    jitter: bool              # True (AWS Full Jitter via tenacity)

def load_config(path: str | Path) -> OrchestratorConfig: ...
def load_config_from_dict(data: dict) -> OrchestratorConfig: ...
def load_config_from_string(yaml_str: str) -> OrchestratorConfig: ...
def get_default_config() -> OrchestratorConfig: ...

class OrchestratorState(TypedDict, total=False):
    """LangGraph state. Uses _merge_dicts reducer for context_slots and knowledge_context."""
    messages: Annotated[list[BaseMessage], add_messages]
    current_agent: str
    context_slots: Annotated[dict[str, Any], _merge_dicts]
    knowledge_context: Annotated[dict[str, Any], _merge_dicts]
    session_id: str
    user_id: str
    request_id: str

@dataclass(frozen=True, slots=True)
class AgentResponse:
    agent_name: str
    content: str
    next_agent: str | None = None
    slots_update: dict[str, Any] = field(default_factory=dict)
    slots_tier: dict[str, SlotTier] = field(default_factory=dict)
    token_usage: dict[str, int] | None = None

class TokenCounter(ABC):
    @abstractmethod
    def count(self, text: str) -> int: ...

class CharDivisionTokenCounter(TokenCounter):
    """max(1, len(text) // chars_per_token) for non-empty; 0 for empty. Default chars_per_token=4."""

class RateLimiter(ABC):
    async def acquire(self) -> None: ...
    async def release(self) -> None: ...  # concrete no-op default
    async def close(self) -> None: ...
    # Also supports async with protocol

class SemaphoreRateLimiter(RateLimiter):
    def __init__(self, *, max_concurrency: int = 1, delay_seconds: float = 0.0): ...

class PerTenantRateLimiter(RateLimiter):
    def __init__(self, *, max_concurrency: int = 1, delay_seconds: float = 0.0, max_tenants: int = 10_000): ...
    def set_tenant(self, tenant_id: str) -> None: ...

class OrchestratorContext:
    """Manages coordinated startup/shutdown of SlotManager, SessionManager,
    ReactTemplateManager, RateLimiter, and extras. Supports async with."""
    async def close(self) -> None: ...
    @property
    def is_closed(self) -> bool: ...

def create_initial_state(
    user_input: str,
    *,
    current_agent: str = "supervisor",
    session_id: str = "default",
    user_id: str = "anonymous",
) -> OrchestratorState: ...

def bind_logging_context(state, config, *, redact_pii=False) -> None: ...
def clear_logging_context() -> None: ...

async def ainvoke_with_timeout(llm, messages, timeout_seconds, config) -> Any: ...
async def astream_with_timeout(llm, messages, timeout_seconds, config) -> AsyncIterator: ...
def build_async_retryer(retry_config: RetryConfig) -> AsyncRetrying: ...

def register_transient_exception(exc_type: type[Exception]) -> None: ...
def deregister_transient_exception(exc_type: type[Exception]) -> None: ...
def get_transient_exceptions() -> tuple[type[Exception], ...]: ...

class BaseAgent(ABC):
    def __init__(
        self,
        config: AgentConfig,
        template_manager: ReactTemplateManager | None = None,
        react_executor: ReactExecutor | None = None,
        middleware: Sequence[AgentMiddleware] | None = None,
        *,
        redact_pii: bool = False,
    ) -> None: ...

    @abstractmethod
    async def process(self, state: OrchestratorState, config: RunnableConfig | None = None) -> AgentResponse: ...
    async def astream_process(self, state, config) -> AsyncGenerator[tuple[dict, str], None]: ...
    async def invoke(self, state: OrchestratorState, config: RunnableConfig | None = None) -> dict[str, Any]: ...

class AgentMiddleware:
    """Override only what you need. All hooks have concrete no-op defaults."""
    async def before_invoke(self, agent_name, state, config) -> OrchestratorState: ...
    async def after_invoke(self, agent_name, state, response, config) -> AgentResponse: ...
    async def on_error(self, agent_name, state, error, config) -> None: ...

class SupervisorAgent(BaseAgent):
    """Built-in supervisor. Simple mode (single LLM call) or ReAct mode."""
    def __init__(self, config, llm, topology, *, template_manager=None,
                 react_executor=None, middleware=None, rate_limiter=None,
                 use_structured_output=False, redact_pii=False): ...
    async def process(self, state, config) -> AgentResponse: ...
    async def astream_process(self, state, config) -> AsyncGenerator: ...

class SessionManagerAgent(BaseAgent):
    """ReAct agent using SessionManager methods as action handlers."""
    def __init__(self, config, llm, session_manager, *,
                 template_manager=None, middleware=None,
                 rate_limiter=None, redact_pii=False): ...

class MetricsMiddleware(AgentMiddleware):
    def __init__(self, collector: MetricsCollector): ...

class TenantIsolationMiddleware(AgentMiddleware):
    """Raises TenantIsolationError if process() mutates session_id/user_id."""

class CircuitBreakerMiddleware(AgentMiddleware):
    """CLOSED -> OPEN -> HALF_OPEN -> CLOSED state machine."""
    def __init__(self, *, failure_threshold: int = 5, recovery_timeout_seconds: float = 30.0): ...

class MemoryInjectionMiddleware(AgentMiddleware):
    """Injects relevant memories into state['knowledge_context']['_injected_memories']."""
    def __init__(self, session_manager: SessionManager, *, limit: int = 5): ...

class SlotTier(StrEnum):
    TURN = "turn"           # Discarded at end of turn
    SESSION = "session"     # Persisted for the session
    LONG_TERM = "long_term" # Persisted across sessions

TIER_ORDER: tuple[SlotTier, ...] = (SlotTier.TURN, SlotTier.SESSION, SlotTier.LONG_TERM)

@dataclass(slots=True)
class Slot:
    key: str
    value: Any
    priority: int
    tier: SlotTier
    created_at: float
    accessed_at: float
    tokens: int

class SlotPool:
    """Token-budgeted collection. Supports len() and 'in' operators."""
    def add(self, key, value, priority=0, *, tier=SlotTier.TURN) -> None: ...
    def get(self, key) -> Any | None: ...          # Updates accessed_at
    def get_slot(self, key) -> Slot | None: ...     # Updates accessed_at
    def remove(self, key) -> bool: ...
    def slots(self) -> list[Slot]: ...
    @property
    def total_tokens(self) -> int: ...

class SlotManager:
    def __init__(self, config: ContextConfig, persistence=None, token_counter=None, timeout_config=None): ...
    def set_shared(self, key, value, priority=0) -> None: ...
    def get_shared(self, key) -> Any | None: ...
    def set_shared_tiered(self, key, value, tier: SlotTier, priority=0) -> None: ...
    def promote_slot(self, key, target_tier: SlotTier) -> bool: ...
    def set_agent(self, agent_name, key, value, priority=0) -> None: ...
    def get_agent(self, agent_name, key) -> Any | None: ...
    async def hydrate(self, session_id, user_id=None) -> None: ...
    async def flush(self, session_id, user_id=None) -> None: ...
    async def clear_turn_slots(self) -> None: ...
    def to_dict(self) -> dict: ...
    def from_dict(self, data: dict) -> None: ...
    async def close(self) -> None: ...
    # Supports async with

class EvictionStrategy(ABC):
    @abstractmethod
    def select(self, slots: list[Slot], count: int) -> list[Slot]: ...

class SlotPersistence(ABC):
    @abstractmethod
    async def load_session_slots(self, session_id: str) -> list[Slot]: ...
    @abstractmethod
    async def save_session_slots(self, session_id: str, slots: list[Slot]) -> None: ...
    @abstractmethod
    async def load_long_term_slots(self, user_id: str) -> list[Slot]: ...
    @abstractmethod
    async def save_long_term_slots(self, user_id: str, slots: list[Slot]) -> None: ...
    async def health_check(self) -> dict[str, Any]: ...

class SessionManager(ABC):
    """3 abstract methods. Everything else raises NotImplementedError by default."""

    # Abstract (must implement)
    @abstractmethod
    async def save_messages(self, session_id: str, messages: list[SessionMessage]) -> None: ...
    @abstractmethod
    async def get_messages(self, session_id: str, limit: int = 50, user_id: str | None = None) -> list[SessionMessage]: ...
    @abstractmethod
    async def clear_session(self, session_id: str) -> None: ...

    # Concrete defaults (override to enable)
    async def save_summary(self, session_id, summary) -> None: ...         # raises NotImplementedError
    async def get_summary(self, session_id) -> Summary | None: ...         # raises NotImplementedError
    async def store_memory(self, session_id, memory, user_id=None) -> None: ...
    async def recall_memory(self, session_id, key) -> Memory | None: ...
    async def search_memories(self, session_id, query, limit=5) -> list[Memory]: ...
    async def search_memories_by_user(self, user_id, query, limit=10) -> list[Memory]: ...
    async def get_messages_paginated(self, session_id, cursor=None, limit=50, user_id=None) -> PaginatedResult: ...
    async def update_memory(self, session_id, memory_id, updates, user_id=None) -> Memory | None: ...
    async def delete_memory(self, session_id, memory_id, user_id=None) -> bool: ...
    async def list_memories(self, session_id, cursor=None, limit=20) -> PaginatedResult[Memory]: ...

    # Utility
    async def health_check(self) -> dict[str, Any]: ...
    def get_capabilities(self) -> set[str]: ...
    async def close(self) -> None: ...
    # Supports async with

class InMemorySessionManager(SessionManager):
    """Full implementation for dev/testing. Per-session asyncio.Lock guards."""

class SemanticSearchMixin(ABC):
    """Mix-in for vector-based memory search. Must appear first in MRO."""
    @abstractmethod
    async def embed_text(self, text: str) -> list[float]: ...
    async def search_memories_semantic(self, session_id, query, limit=5) -> list[Memory]: ...

class SessionMessage(BaseModel):
    id: str                    # UUID
    session_id: str
    role: Literal["user", "assistant", "ai", "system"]
    content: str
    timestamp: datetime
    user_id: str | None = None
    metadata: dict[str, Any] = {}
    def to_langchain(self) -> BaseMessage: ...

class Memory(BaseModel):
    id: str
    session_id: str
    key: str
    value: str
    importance: float          # ge=0.0, le=1.0
    user_id: str | None = None
    expires_at: datetime | None = None
    # ... categories, metadata, source_agent, etc.

class Summary(BaseModel):
    id: str
    session_id: str
    content: str
    user_id: str | None = None

class PaginatedResult(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None
    has_more: bool

class TopologyResolver:
    """Validates allowed routes from topology config."""
    def __init__(self, config: TopologyConfig): ...
    def get_allowed_targets(self, from_agent: str) -> list[str]: ...
    @property
    def default_return(self) -> str: ...

def build_conditional_edge(
    topology: TopologyResolver, from_agent_name: str
) -> Callable[[OrchestratorState], str]:
    """Returns a routing function for StateGraph.add_conditional_edges().
    Returns END when current_agent == from_agent_name (termination sentinel).
    Falls back to default_return for unknown targets."""

def build_hub_spoke_graph(
    supervisor, agents: dict[str, Any], topology: TopologyResolver,
    *, state_schema=OrchestratorState
) -> CompiledStateGraph:
    """Hub-and-spoke: all workers route back to supervisor."""

def build_topology_graph(
    agents: dict[str, Any], topology: TopologyResolver,
    *, entry_point: str, state_schema=OrchestratorState
) -> CompiledStateGraph:
    """Flexible: edges match topology exactly. Supports mesh, hybrid."""

class ReactConfigSchema(BaseModel):
    """frozen=True, extra='forbid'. Validates react_config.yaml."""
    max_passes: int = DEFAULT_MAX_PASSES  # default 5, gt=0
    actions: dict[str, ReactActionSchema]
    slots: dict[str, ReactSlotSchema]
    # @model_validator: validates dependency references + cycle detection (DFS)

class ReactSlotSchema(BaseModel):
    description: str
    modules: list[str]         # min_length=1, relative .md file paths
    dependencies: list[str]    # slot names that must load first
    triggers: str              # hint for when to load
    max_tokens: int | None     # per-slot token budget cap

class ReactActionSchema(BaseModel):
    description: str
    parameters: dict[str, str]  # param name -> description
    required: list[str]

DEFAULT_MAX_PASSES: int = 5

class KnowledgeLoader(ABC):
    @abstractmethod
    async def load_module(self, module_ref: str) -> str: ...
    @abstractmethod
    def resolve_module_path(self, module_ref: str) -> Path: ...

class FileSystemKnowledgeLoader(KnowledgeLoader):
    """Enforces base_dir containment. Rejects absolute paths and traversal."""

class ReactTemplateManager:
    """Per-agent config + slot loading. Async-safe with asyncio.Lock."""
    @classmethod
    async def create(cls, ...) -> ReactTemplateManager: ...  # async factory
    async def load_slot(self, slot_name: str) -> KnowledgeLoadResult: ...
    async def load_core_prompt(self, agent_name, **kwargs) -> str: ...
    def format_available_slots(self) -> str: ...
    def format_available_actions(self) -> str: ...
    async def reset(self) -> None: ...
    async def close(self) -> None: ...
    @property
    def max_passes(self) -> int: ...
    @property
    def token_counter(self) -> TokenCounter: ...
    # Supports async with

class ReactExecutor:
    def __init__(
        self,
        agent_name: str,
        llm: BaseLanguageModel,
        template_manager: ReactTemplateManager,
        action_handlers: dict[str, ActionHandler],
        *,
        max_passes: int | None = None,
        retry_config: RetryConfig | None = None,
        timeout_config: TimeoutConfig | None = None,
        max_consecutive_errors: int | None = None,
        max_history_tokens: int | None = None,
        rate_limiter: RateLimiter | None = None,
        cancel_event: asyncio.Event | None = None,
        use_structured_output: bool = False,
        total_request_seconds: float | None = None,
    ) -> None: ...

    async def execute(
        self, user_input: str, context: dict[str, Any],
        config: RunnableConfig | None = None, *, request_id: str | None = None
    ) -> dict[str, Any]: ...

class ReactParsedResponse:
    thought: str
    action: str
    action_params: dict[str, Any]

class ReactStructuredResponse(BaseModel):
    """Pydantic model for llm.with_structured_output()."""
    thought: str
    action: str
    action_params: dict[str, Any]

class TerminalActionResult(Exception):
    """Raised by action handlers to signal loop termination."""
    def __init__(self, result: dict[str, Any]): ...

# ActionHandler is a Protocol with optional config param
ActionHandler = Callable[[dict, dict, RunnableConfig | None], Awaitable[tuple[str, dict]]]

OTEL_AVAILABLE: bool  # True if opentelemetry-api is importable

def get_tracer(name: str = "graphcrew") -> Tracer: ...
async def otel_span(name: str, **attributes) -> AsyncContextManager: ...
def sync_otel_span(name: str, **attributes) -> ContextManager: ...

class OpenTelemetryMiddleware(AgentMiddleware):
    """Creates spans for agent invocations. Per-invocation state in contextvars."""

class MockLLM:
    """Returns scripted responses without real API calls."""
    # Supports:
    #   raise_on_call: exception class to raise
    #   stream_chunks: per-call chunk lists for astream
    #   structured_responses: FIFO list for with_structured_output
    #   usage_tokens: dict attached to every response

def make_test_state(user_input: str, **kwargs) -> OrchestratorState:
    """Creates a minimal OrchestratorState for testing."""

# Re-exported for convenience:
InMemorySessionManager

Exception
  |
  +-- OrchestratorError                          # Base for all library exceptions
       |
       +-- ConfigurationError                    # Base for config/setup errors
       |    +-- LLMNotConfiguredError            # Agent requires LLM but none provided
       |    +-- ReactExecutorNotConfiguredError   # ReAct mode needs template_manager
       |    +-- ConfigLoadError*                  # Config file not found or unreadable
       |    +-- ConfigValidationError*            # Config YAML fails Pydantic validation
       |
       +-- AgentExecutionError                   # Base for agent runtime failures
       |    +-- AgentProcessError                # Agent's process() method raised
       |
       +-- LLMError                              # Base for LLM call failures
       |    +-- RateLimitExceededError            # Rate limiter rejected the call
       |    +-- LLMInvocationError               # Single LLM ainvoke() call failed
       |    +-- LLMRetryExhaustedError           # All retry attempts exhausted
       |
       +-- SessionError                          # Base for session layer errors
       |    +-- SessionManagerError              # SessionManager operation failed
       |    +-- SessionManagerNotConfiguredError  # No session manager provided
       |
       +-- StateError                            # Base for state validation errors
       |    +-- InvalidStateError                # Required state key missing
       |
       +-- ContextError                          # Base for slot/context errors
       |    +-- SlotPersistenceError             # Persistence load/save failed
       |
       +-- KnowledgeError                        # Base for knowledge module errors
       |    +-- KnowledgeSlotNotFoundError       # Requested slot does not exist
       |    +-- KnowledgePathError               # Module path escapes base directory
       |    +-- KnowledgeLoadError               # Module file failed to load
       |    +-- KnowledgeConfigValidationError   # react_config.yaml validation failed
       |
       +-- ReactError                            # Base for ReAct execution errors
       |    +-- ReactCancellationError           # ReAct loop cancelled via asyncio.Event
       |    +-- ReactMaxPassesError              # max_passes exceeded without terminal action
       |    +-- ReservedStateKeyError            # Attempted overwrite of reserved state keys
       |    +-- ReactConsecutiveErrorsError      # Max consecutive invalid actions reached
       |
       +-- RoutingError                          # Base for routing errors
       |    +-- RoutingSelfLoopError             # Agent tried to route to itself
       |
       +-- TenantIsolationError                  # Agent mutated session_id/user_id
       |
       +-- CircuitBreakerOpenError               # Circuit breaker rejecting calls

tests/
  conftest.py                          # Global fixtures (autouse structlog cleanup)
  __init__.py
  helpers/
    __init__.py
    knowledge_helpers.py               # setup_supervisor_agent_dir(), setup_minimal_agent_dir()
    session_helpers.py                 # FullSessionManager, make_session_agent_config()
  unit/
    __init__.py
    config/
      test_schema.py                   # Config model validation
      test_loader.py                   # YAML loading, ConfigLoadError
      test_defaults.py                 # Default value verification
      test_agent_name_property.py      # Hypothesis property-based name validation
      test_retry_config.py             # Retry/backoff config
      test_timeout_config.py           # Timeout config validation
    core/
      test_state.py                    # OrchestratorState, create_initial_state()
      test_types.py                    # Type aliases and enums
      test_exceptions.py               # Exception hierarchy
      test_logging_context.py          # bind_logging_context, PII redaction
      test_token_counting.py           # CharDivisionTokenCounter + hypothesis
      test_llm_utils.py                # Retry, transient exceptions, timeouts
      test_rate_limiting.py            # SemaphoreRateLimiter, PerTenantRateLimiter
      test_orchestrator_context.py     # OrchestratorContext lifecycle
    agents/
      test_base.py                     # BaseAgent ABC, middleware execution
      test_supervisor.py               # SupervisorAgent routing
      test_session_manager_agent.py    # SessionManagerAgent capabilities
      test_middleware.py               # AgentMiddleware onion model
      test_metrics.py                  # MetricsMiddleware, InMemoryMetricsCollector
      test_tenant_isolation.py         # TenantIsolationMiddleware
      test_circuit_breaker.py          # CircuitBreakerMiddleware state machine
      test_memory_injection.py         # MemoryInjectionMiddleware
      test_streaming.py                # astream_process()
    context/
      test_slots.py                    # SlotPool, SlotTier + hypothesis
      test_manager.py                  # SlotManager lifecycle
      test_eviction.py                 # FIFO, LRU, Priority eviction
      test_persistence.py              # InMemorySlotPersistence
    session/
      test_interface.py                # SessionManager ABC
      test_in_memory.py                # InMemorySessionManager
      test_models.py                   # SessionMessage, Memory, Summary + hypothesis
      test_semantic.py                 # SemanticSearchMixin
    routing/
      test_topology.py                 # TopologyResolver
      test_graph_builder.py            # build_hub_spoke_graph(), build_conditional_edge()
    knowledge/
      test_loader.py                   # FileSystemKnowledgeLoader
      test_template_manager.py         # ReactTemplateManager lifecycle
      test_models.py                   # ReactActionSchema, ReactSlotSchema
      test_react_config_schema.py      # react_config.yaml validation
    react/
      test_executor.py                 # ReactExecutor multi-pass loop
      test_structured_output.py        # ReactStructuredResponse, fallback
      test_models.py                   # ReAct response models
    observability/
      test_otel.py                     # OpenTelemetry middleware (optional)
    test_concurrency.py                # asyncio.Lock guards across components
    test_testing_subpackage.py         # MockLLM, make_test_state validation
    test_public_api.py                 # __all__ exports, __version__
    test_subpackage_exports.py         # Subpackage re-exports
    test_chat.py                       # run_chat_turn()
  integration/
    test_agent_e2e.py                  # End-to-end agent invocation
    test_react_integration.py          # Full ReAct loop with MockLLM
    test_session_integration.py        # Session persistence round-trips
    test_example_chatbot.py            # Smoke test for simple_chatbot example
    test_example_react_knowledge.py    # Smoke test for react_knowledge_agent example
    test_graph_execution.py            # Compiled LangGraph execution
    test_multi_tenant.py               # Multi-tenant isolation, cross-session memory
  performance/
    __init__.py
    test_benchmarks.py                 # Opt-in benchmarks

from graphcrew.testing import MockLLM, make_test_state, InMemorySessionManager

"""Tests for the FooBar component."""

from __future__ import annotations

import pytest
from pydantic import ValidationError

from graphcrew.config.schema import AgentConfig


class TestAgentConfig:
    def test_valid_config(self) -> None:
        cfg = AgentConfig(name="supervisor", type="supervisor", description="Main")
        assert cfg.name == "supervisor"

    def test_missing_required_field_raises(self) -> None:
        with pytest.raises(ValidationError):
            AgentConfig(type="custom", description="No name")  # type: ignore[call-arg]

    @pytest.mark.parametrize(
        ("name", "match"),
        [
            ("", None),
            ("__end__", "reserved"),
            ("1agent", "invalid"),
        ],
        ids=["empty", "reserved-end", "digit-start"],
    )
    def test_invalid_names(self, name: str, match: str | None) -> None:
        with pytest.raises(ValidationError):
            AgentConfig(name=name, type="custom", description="desc")

# Full test suite with coverage
uv run pytest --cov=graphcrew --cov-report=term-missing --cov-fail-under=90

# Via Make
make test

# Specific test file
uv run pytest tests/unit/agents/test_supervisor.py -v

from __future__ import annotations

model_config = ConfigDict(frozen=True, extra="forbid")

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_configs = true

git clone https://github.com/Atiqul-Islam/swagent.git
cd swagent
uv sync --group dev
pre-commit install

make test       # Run tests with coverage (fail-under=90)
make lint       # Run ruff linter
make typecheck  # Run mypy strict type checking
make format     # Auto-format code with ruff
make audit      # Run pip-audit vulnerability scan
make all        # Run lint + typecheck + audit + test
make clean      # Remove build artifacts, caches

[build-system]
requires = ["hatchling>=1.25"]
build-backend = "hatchling.build"

[project.optional-dependencies]
otel = ["opentelemetry-api>=1.24,<2.0"]