observable-agent-loop-solution/main.py at main · Kunal-CodeLab/observable-agent-loop-solution · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
"""
Main entry point for Caygnus Observable Agent Loop.
Run pre-configured acceptance scenarios or interactive incident investigations.
"""

from __future__ import annotations
import argparse
import sys

from src.agent import AgentLoop
from src.tools import create_default_registry, FailingToolMock
from src.model import IncidentInvestigatorDynamicAdapter, ScriptedDeterministicAdapter
from src.cli import print_banner, render_trace
from src.types import ModelDecision, ModelDecisionType, ToolCall, StructuredAgentResponse, LoopStatus, EvidenceItem


def run_scenario(scenario_name: str, output_json: bool = False):
    print_banner()

    if scenario_name == "single-step":
        print("\n>>> Running Scenario: Single-Step Tool Selection (AC1) <<<")
        registry = create_default_registry()
        loop = AgentLoop(
            model_adapter=IncidentInvestigatorDynamicAdapter(),
            tool_registry=registry,
            max_steps=5,
        )
        objective = "Check the current health and status of payment-service."
        trace = loop.run(objective)
        render_trace(trace, output_json=output_json)

    elif scenario_name == "multi-step":
        print("\n>>> Running Scenario: Multi-Step Investigation & Evidence Synthesis (AC2, AC3, AC6) <<<")
        registry = create_default_registry()
        loop = AgentLoop(
            model_adapter=IncidentInvestigatorDynamicAdapter(),
            tool_registry=registry,
            max_steps=6,
        )
        objective = "Investigate elevated 500 error rates on payment-service, correlate with metrics/logs, and recommend remediation."
        trace = loop.run(objective)
        render_trace(trace, output_json=output_json)

    elif scenario_name == "tool-failure":
        print("\n>>> Running Scenario: Tool Failure & Adaptive Recovery (AC4) <<<")
        registry = create_default_registry()
        # Register a failing diagnostic probe
        registry.register(FailingToolMock(failure_count=1))
        loop = AgentLoop(
            model_adapter=IncidentInvestigatorDynamicAdapter(),
            tool_registry=registry,
            max_steps=5,
        )
        objective = "Probe external network endpoint for payment failures, and fallback to application logs if probe fails."
        trace = loop.run(objective)
        render_trace(trace, output_json=output_json)

    elif scenario_name == "step-limit":
        print("\n>>> Running Scenario: Configured Step Limit Enforcement (AC5) <<<")
        registry = create_default_registry()
        # Set a tight max_steps=3 on an open-ended loop
        loop = AgentLoop(
            model_adapter=IncidentInvestigatorDynamicAdapter(),
            tool_registry=registry,
            max_steps=3,
        )
        objective = "Run infinite loop test across dependencies."
        trace = loop.run(objective)
        render_trace(trace, output_json=output_json)

    elif scenario_name == "interactive":
        print("\n>>> Interactive Investigation Mode <<<")
        user_obj = input("Enter incident objective: ").strip()
        if not user_obj:
            user_obj = "Investigate payment-service degradation"
        registry = create_default_registry()
        loop = AgentLoop(
            model_adapter=IncidentInvestigatorDynamicAdapter(),
            tool_registry=registry,
            max_steps=5,
        )
        trace = loop.run(user_obj)
        render_trace(trace, output_json=output_json)

    else:
        print(f"Unknown scenario '{scenario_name}'. Available: single-step, multi-step, tool-failure, step-limit, interactive")
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="Caygnus Observable Agent Loop - Incident Investigator")
    parser.add_argument(
        "--scenario",
        type=str,
        default="multi-step",
        choices=["multi-step", "single-step", "tool-failure", "step-limit", "interactive"],
        help="Pre-configured evaluation scenario to run (default: multi-step)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output raw JSON summary instead of formatted terminal UI",
    )
    args = parser.parse_args()
    run_scenario(args.scenario, output_json=args.json)


if __name__ == "__main__":
    main()