Observability with shipit-watcher

# Setup — run once, BEFORE anything else.
#
# Order matters: the repo checkouts go on sys.path first, so an older
# `shipit_agent` already installed in this kernel cannot win the import.
# Importing it before adjusting the path caches the wrong copy, and
# WatcherExporter then appears to be missing.
import subprocess
import sys
import pathlib

repo = pathlib.Path.cwd().parent                 # .../shipit_agent
watcher = repo.parent / "shipit-watcher"

for path in (str(watcher), str(repo)):
    if path not in sys.path:
        sys.path.insert(0, path)

# langfuse is a real dependency; the two repos are used from source.
try:
    import langfuse
except ImportError:
    subprocess.run([sys.executable, "-m", "pip", "install", "-q",
                    "langfuse==2.60.10"], check=True)

if not watcher.exists():
    # No sibling checkout — fall back to the published package.
    subprocess.run([sys.executable, "-m", "pip", "install", "-q",
                    "shipit-watcher"], check=True)

import shipit_agent
import shipit_watcher
from shipit_agent.tracing_exporters import WatcherExporter

print(f"shipit-agent   {shipit_agent.__version__}")
print(f"               {shipit_agent.__file__}")
print(f"shipit-watcher {pathlib.Path(shipit_watcher.__file__).parent}")
print("WatcherExporter: ready")

import os
import pathlib

# Load .env without a dependency.
for line in pathlib.Path("../.env").read_text().splitlines():
    line = line.strip()
    if line and not line.startswith("#") and "=" in line:
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip())

from shipit_watcher.config import WatcherConfig

c = WatcherConfig()
print(f"host        : {c.langfuse_host}")
print(f"public key  : {'set' if c.langfuse_public_key else 'MISSING'}")
print(f"secret key  : {'set' if c.langfuse_secret_key else 'MISSING'}")
print(f"transport   : {c.langfuse_transport}")
print(f"service/env : {c.service_name} / {c.environment}")
print(f"masking     : {c.mask_pii}   bodies sent: {c.capture_content}")
print(f"governance  : {c.governance_mode}")
print(f"local ledger: {c.persist_to_database}  -> {c.ledger_model}")
print()
print(f"credentials ok: {c.has_langfuse_credentials}   emitting: {c.is_active}")

from shipit_watcher.sinks.langfuse_otel_sink import LangfuseOTLPSink

sink = LangfuseOTLPSink()
print("OTLP endpoint :", sink._endpoint or "(no host configured)")
print("auth header   :", "set" if sink._auth else "MISSING")
print("available     :", sink.available)

from shipit_agent import Agent

exporter = WatcherExporter()
# agent = Agent(llm=llm, tools=tools, trace_store=exporter)
print(type(exporter).__name__, "is a TraceStore:",
      hasattr(exporter, "append_event") and hasattr(exporter, "load"))

from shipit_watcher.tracer import Tracer
from shipit_agent.llms.base import LLMResponse
from shipit_agent.models import ToolCall
from shipit_agent.tools.base import ToolOutput

seen = []

class SpySink:
    def start_trace(self, trace_id, name, context, input_data=None):
        seen.append(("trace opened", name))
    def end_trace(self, trace_id, output=None, metadata=None):
        seen.append(("trace closed", ""))
    def record(self, event, context):
        seen.append((event.type.value, event.name))
    def record_score(self, score):
        seen.append(("score", f"{score.name}={score.value}"))
    def flush(self):
        pass

class ScriptedLLM:
    model = "demo"
    def __init__(self): self.n = 0
    def complete(self, **kw):
        self.n += 1
        if self.n == 1:
            return LLMResponse(content="Reading the config first.",
                               tool_calls=[ToolCall(name="read_file",
                                                    arguments={"path": "app.py"})])
        return LLMResponse(content="app.py sets DEBUG=True.")

class ReadFile:
    name, description, prompt_instructions = "read_file", "Read a file", ""
    def schema(self):
        return {"function": {"name": "read_file", "parameters":
                {"properties": {"path": {"type": "string"}}}}}
    def run(self, context, **kw):
        return ToolOutput(text="DEBUG = True",
                          metadata={"summary": "12 bytes"})

result = Agent(
    llm=ScriptedLLM(), tools=[ReadFile()],
    trace_store=WatcherExporter(tracer=Tracer(sinks=[SpySink()])),
    progress_summaries=True, auto_use_skills=False, max_iterations=4,
).run("what does app.py set?")

for kind, name in seen:
    print(f"  {kind:16} {name}")

for event in result.events:
    if event.type in ("agent_decision", "agent_observation"):
        print(f"  {event.type:18} {event.payload.get('summary')}")

if c.is_active:
    live = WatcherExporter()
    Agent(llm=ScriptedLLM(), tools=[ReadFile()], trace_store=live,
          progress_summaries=True, auto_use_skills=False,
          max_iterations=4).run("what does app.py set?")
    live.flush()
    print("sent — open", c.langfuse_host, "and look for service =", c.service_name)
else:
    print("skipped: no credentials. Fill LANGFUSE_* in .env")

if c.is_active:
    from shipit_watcher import Score, record_score
    record_score(Score(name="helpfulness", value=0.9,
                       comment="answered from the file, no guessing"))
    print("score sent")
else:
    print("skipped: no credentials")

if c.is_active:
    from shipit_watcher import capture, create_dataset
    create_dataset("agent-regressions", description="cases that went wrong")
    capture("agent-regressions",
            input="what does app.py set?",
            expected_output="DEBUG = True")
    print("captured")
else:
    print("skipped: no credentials")

print(f"persist_to_database : {c.persist_to_database}")
print(f"ledger model        : {c.ledger_model}")
print(f"event model         : {c.ledger_event_model}")
if not c.persist_to_database:
    print("\n-> nothing is written locally. If the tunnel drops, those runs are gone.")