diff --git a/README.md b/README.md index e5a34ef..3b73223 100644 --- a/README.md +++ b/README.md @@ -277,6 +277,7 @@ See the [Getting Started guide](docs/getting-started.md) for more options. | **Web Dashboard** | Browser-based session explorer with activity timeline (swim-lane visualization), step list, context window viewer, visual timeline diff, multi-metric axis (duration/tokens/cost), and live recording via WebSocket. Everything embedded in the single binary. | [web-ui.md](docs/web-ui.md) | - | | **Multi-Agent Tracing** | Hierarchical span tree and activity timeline for multi-agent workflows. Each agent gets its own swim lane with duration bars. Auto-captures agent boundaries, tool calls, and handoffs from OpenAI Agents SDK. Manual `@span()` decorator for custom grouping. Thread view for multi-turn conversations. | [multi-agent-tracing.md](docs/multi-agent-tracing.md) | - | | **Framework Integrations** | Native support for OpenAI Agents SDK and Pydantic AI (auto-detected on `init()`). Wrapper support for LangGraph and CrewAI. Any other framework works via the HTTP proxy. | [framework-integrations.md](docs/framework-integrations.md) | [09_pydantic_ai.py](examples/09_pydantic_ai.py), [10_openai_agents_sdk.py](examples/10_openai_agents_sdk.py) | +| **Bring your own framework** | One-call `rewind_agent.connector.setup()` for any agent on httpx/requests/aiohttp — wraps a session and HTTP intercept in a single `with` block. Custom transport (gRPC, in-process)? Use `ExplicitClient` directly. Three-tier guide covers all paths. | [hdk.md](docs/hdk.md) | - | | **Claude Code Observation** | Observe Claude Code sessions in real-time via hooks. See every tool call (Read, Edit, Bash, Grep, Agent), user prompts, and session lifecycle. Token usage extracted from transcripts. One-command setup: `rewind hooks install`. | - | - | | **MCP Server** | 26 tools for AI assistants (Claude Code, Cursor, Windsurf) to query recordings, view span trees, browse threads, diff timelines, create baselines, run evals - all from your IDE. | [mcp-server.md](docs/mcp-server.md) | - | | **OpenTelemetry Export** | Export recorded sessions as OTel traces via OTLP to Langfuse, Datadog, Grafana Tempo, Jaeger, or any OTel-compatible backend. CLI, Python SDK, and Web API. Uses `gen_ai.*` semantic conventions. Privacy-first: message content requires explicit opt-in. | [otel-export.md](docs/otel-export.md) | - | @@ -303,6 +304,7 @@ See the [Getting Started guide](docs/getting-started.md) for more options. |:------|:-----------|:--------------| | **Native** - auto-detected on `init()` | [OpenAI Agents SDK](https://github.com/openai/openai-agents-python), [Pydantic AI](https://ai.pydantic.dev/) | Zero config. Agent boundaries, tool calls, and handoffs captured automatically. | | **Wrapper** - manual setup | LangGraph, CrewAI | Thin integration via `wrap_langgraph()` / `wrap_crew()`. CrewAI requires proxy mode. | +| **Bring your own** | Custom agent / framework / org connector | One-liner `rewind_agent.connector.setup(name=...)` for HTTP-shaped LLM clients; `ExplicitClient` for custom transports. See [hdk.md](docs/hdk.md). | | **Works via proxy** | Any framework using OpenAI/Anthropic APIs | Point `OPENAI_BASE_URL` at the proxy. Works with Autogen, smolagents, custom code, any language. | ## Works With Your Observability Stack diff --git a/docs/framework-integrations.md b/docs/framework-integrations.md index 5365739..a91ca0c 100644 --- a/docs/framework-integrations.md +++ b/docs/framework-integrations.md @@ -85,6 +85,22 @@ crew = rewind_agent.wrap_crew(crew) result = crew.kickoff() ``` +## Bring your own framework + +Building an agent or framework that isn't in the list above? Don't write a custom HTTP emitter — Rewind ships the integration primitives. Three tiers, pick the one that matches your transport: + +- **Tier 1** — your agent uses `httpx` / `requests` / `aiohttp`: one-liner via `rewind_agent.connector.setup()`. +- **Tier 2** — custom transport (gRPC, in-process, mTLS-tunneled): ~30-line `ExplicitClient` integration. +- **Tier 3** — shipping a connector for an org or framework family (one package consumed by every internal agent): thin wrapper around Tier 1 with org-specific defaults. + +```python +import rewind_agent +with rewind_agent.connector.setup(name="my-agent"): + run_agent_loop() +``` + +See **[hdk.md](hdk.md)** for the full decision tree, configuration knobs (`REWIND_ENABLED`, `REWIND_URL`, `REWIND_LLM_HOSTS`), and common pitfalls. + ## Examples - [`examples/04_langgraph.py`](../examples/04_langgraph.py) -- LangGraph integration diff --git a/docs/hdk.md b/docs/hdk.md new file mode 100644 index 0000000..8757ec2 --- /dev/null +++ b/docs/hdk.md @@ -0,0 +1,215 @@ +# Integrate any framework with Rewind + +You're building an agent — or shipping a framework that agents use — and you want recording, fork, and replay without writing a custom HTTP emitter for every app. This page is the decision tree. + +The big idea: **Rewind already ships the parts.** You don't need a framework adapter, a Protocol, or a plugin. Pick the tier that matches your transport and call site, write 5–35 lines, done. + +## Pick a tier + +``` +Are your LLM calls HTTP-shaped? (httpx / requests / aiohttp) +├── Yes → Tier 1: intercept.install() ~5 LOC +└── No → Tier 2: ExplicitClient ~30 LOC + +Need to bake in defaults for your org / framework? + → Tier 3: ship a small connector package ~50 LOC +``` + +| Tier | Use when | Code size | API | +|:---|:---|:---|:---| +| 1 | LLM calls go over httpx / requests / aiohttp (most modern Python LLM clients) | ~5 LOC | [`intercept.install()`](intercept-quickstart.md) | +| 2 | Custom transport: gRPC, in-process LLM, mTLS-tunneled, message bus | ~30 LOC | [`ExplicitClient`](recording-api.md) | +| 3 | You're shipping a connector for an org or framework family (one package consumed by every internal agent) | ~50 LOC | Tier 1 or 2 + a thin wrapper | + +Most integrators stop at Tier 1. + +## Tier 1 — `connector.setup()` + +For any agent that talks to LLMs over httpx, requests, or aiohttp. The connector wraps a session, installs HTTP intercept, and tears both down on exit — one call, one `with` block, no order to remember. + +```python +import rewind_agent + +with rewind_agent.connector.setup(name="my-agent"): + run_agent_loop() +``` + +For a custom LLM gateway hostname, pass `llm_hosts` (or set `REWIND_LLM_HOSTS=a.example,b.example` in the env): + +```python +with rewind_agent.connector.setup( + name="my-agent", + llm_hosts=("llm-gateway.example",), +): + run_agent_loop() +``` + +The yielded value is an `ExplicitClient` — useful for non-HTTP record paths (gRPC, in-process LLMs) inside the same block: + +```python +with rewind_agent.connector.setup(name="my-agent") as client: + response = my_grpc_client.chat(request) + client.record_llm_call( + request=request, response=response.dict(), + model="my-private-llama", duration_ms=duration_ms, + ) +``` + +**Configuration knobs** (env > default; kwargs override env): + +- `REWIND_ENABLED=0` — kill switch with zero overhead in prod when off. +- `REWIND_URL` — Rewind server URL (default `http://127.0.0.1:4800`). +- `REWIND_LLM_HOSTS` — comma-separated hostnames to treat as LLM gateways in addition to the strict-by-default provider list. + +**Replay-aware**: when `REWIND_SESSION_ID` and `REWIND_REPLAY_CONTEXT_ID` are set (the runner-subprocess pattern from [runners.md](runners.md)), `setup()` skips creating a fresh session and lets `intercept.install()` attach to the existing replay context. Drop the connector into runner-driven replay handlers without phantom sessions. + +For per-library examples, custom predicates, streaming behavior, the savings counter, and troubleshooting, see [intercept-quickstart.md](intercept-quickstart.md). + +### When you need finer control + +If you need to install intercept and start sessions independently — for example, a long-running process that opens many short sessions back-to-back — use the underlying primitives directly: + +```python +from rewind_agent.explicit import ExplicitClient +from rewind_agent.intercept import DefaultPredicates, install, uninstall + +class MyPredicates(DefaultPredicates): + def is_llm_call(self, req): + return "llm-gateway.example" in req.url_parts.netloc or super().is_llm_call(req) + +client = ExplicitClient() +install(predicates=MyPredicates()) # once at process startup +try: + for task in tasks: + with client.session(name=task.name): # short session per task + run_task(task) +finally: + uninstall() +``` + +The order matters: the session sets `_session_id` / `_timeline_id` contextvars; the intercept layer reads those vars on every recorded call. Without an active session, intercepted calls silently no-op. **This is the most common Tier-1 mistake** — `connector.setup()` exists specifically to remove this footgun. + +## Tier 2 — `ExplicitClient` + +For agents that don't go over HTTP — gRPC, in-process LLMs, message-bus dispatch, anything `intercept` can't see. You wrap your existing call sites with explicit `record_*` calls. + +```python +import time +from rewind_agent.explicit import ExplicitClient + +client = ExplicitClient() +with client.session("my-agent"): + t0 = time.time() + # your custom LLM call + request = {"messages": [{"role": "user", "content": "hi"}]} + response = my_grpc_client.chat(request) + duration_ms = int((time.time() - t0) * 1000) + + client.record_llm_call( + request=request, + response=response.to_dict(), + model="my-private-llama-70b", + duration_ms=duration_ms, + tokens_in=response.usage.prompt_tokens, + tokens_out=response.usage.completion_tokens, + ) +``` + +`record_tool_call` records tool / function-call steps. `get_replayed_response` checks the replay cache during fork+replay; on a hit, return the cached response instead of calling the live model. The full HTTP wire format and method reference live in [recording-api.md](recording-api.md). + +Async variants (`record_llm_call_async`, `record_tool_call_async`, `session_async`, `get_replayed_response_async`) are drop-in for asyncio code paths and run HTTP in a thread executor so the event loop never blocks. + +## Tier 3 — ship a connector + +If you're integrating a framework family — every internal agent in your org, every team using a private LLM gateway, every agent in a monorepo — package the boilerplate as a small connector. ~50 LOC, one place to update defaults. + +The pattern: + +```python +# my_org_rewind/__init__.py +import os +from contextlib import contextmanager +from rewind_agent.explicit import ExplicitClient +from rewind_agent.intercept import DefaultPredicates, install, uninstall + +_DEFAULT_LLM_HOSTS = ("llm-gateway.your-org.example",) + +class _OrgPredicates(DefaultPredicates): + def __init__(self, hosts): + super().__init__() + self._hosts = tuple(h.strip() for h in hosts if h and h.strip()) + + def is_llm_call(self, req): + if any(h in req.url_parts.netloc for h in self._hosts): + return True + return super().is_llm_call(req) + +@contextmanager +def setup(name, *, base_url=None, llm_hosts=None, enabled=None): + """Connect a my-org-hosted agent to Rewind. + + Override defaults via env (REWIND_ENABLED, REWIND_URL, REWIND_LLM_HOSTS) + or per-call kwargs. Yields the ExplicitClient for non-HTTP record_* calls. + """ + if enabled is None: + enabled = os.environ.get("REWIND_ENABLED", "1") != "0" + if not enabled: + yield None + return + + hosts = llm_hosts + if hosts is None: + env = os.environ.get("REWIND_LLM_HOSTS", "") + hosts = tuple(env.split(",")) if env else _DEFAULT_LLM_HOSTS + + url = base_url or os.environ.get("REWIND_URL", "http://127.0.0.1:4800") + client = ExplicitClient(base_url=url) + with client.session(name): + install(predicates=_OrgPredicates(hosts)) + try: + yield client + finally: + uninstall() +``` + +Every agent in your org becomes a 3-line integration: + +```python +import my_org_rewind +with my_org_rewind.setup("alerts-triage"): + asyncio.run(main()) +``` + +Configurable knobs every connector should expose: + +- `REWIND_ENABLED=0` — kill switch with zero overhead in prod when off. +- `REWIND_URL` — Rewind server URL (default `http://127.0.0.1:4800`). +- `REWIND_LLM_HOSTS` — comma-separated hostnames to treat as LLM gateways. +- Per-call kwargs override env values, useful for tests and multi-tenant cases. + +## Common pitfalls + +**Recording silently no-ops.** `record_llm_call` returns `None` when no session is active. Always wrap your agent loop in `with client.session(...)` before calling `install(...)` or `record_*`. The intercept layer reuses the session's contextvars; without them, intercepted calls succeed (HTTP-wise) but record nothing. + +**Don't combine `init()` with `intercept.install()`.** `rewind_agent.init()` patches the OpenAI / Anthropic Python SDKs at the SDK layer; `intercept.install()` patches httpx / requests / aiohttp at the transport layer. Running both in the same process double-records every call. Pick one path per process. + +**`connector.setup()` is sync, even in async code.** It's a `@contextmanager`, not an `@asynccontextmanager`. Async callers should use plain `with` — not `async with` — even inside an `async def`: + +```python +async def main(): + with rewind_agent.connector.setup(name="my-agent"): # NOT `async with` + await run_agent_loop() +``` + +Only the setup and teardown are synchronous (HTTP POST `/api/sessions/start` and `/api/sessions/{id}/end`); the body of the block can be fully async. + +**`requests.Session()` constructed before `install()` keeps its old adapter.** The intercept layer patches `Session.__init__`; live instances aren't mutated. Move `install()` earlier in startup, or call `session.mount(...)` explicitly for pre-existing sessions. Same caveat applies to `httpx.Client` constructed pre-install. + +**Replay needs a replay context.** Fork-and-replay against a recorded session works automatically when the dispatch flow runs through `runner.py` (see [runners.md](runners.md)). For custom replay drivers, call `client.attach_replay_context(session_id, replay_context_id)` before `install()`; intercept will then serve cached responses on each step. + +## Where to go next + +- [intercept-quickstart.md](intercept-quickstart.md) — full Tier 1 reference: per-library examples, custom predicates, streaming behavior, savings counter, troubleshooting. +- [recording-api.md](recording-api.md) — HTTP wire format and `ExplicitClient` method reference for Tier 2. +- [framework-integrations.md](framework-integrations.md) — first-class adapters for OpenAI Agents SDK, Pydantic AI, LangGraph, CrewAI. +- [runners.md](runners.md) — wiring fork+replay into your runtime. diff --git a/docs/runners.md b/docs/runners.md index 3d0a198..2a4cb45 100644 --- a/docs/runners.md +++ b/docs/runners.md @@ -365,3 +365,12 @@ The runner authenticated successfully but is trying to post events on a job belonging to a different runner. Usually means the runner is processing a stale/leaked dispatch — check runner logs for an old job_id. + +--- + +## Integrating a non-runner agent with Rewind + +Runners are the dashboard's "Run replay" path. If you're starting from a +plain agent and want recording / fork / replay without a webhook, see +[hdk.md](hdk.md) — three-tier guide for picking between +`intercept.install()`, `ExplicitClient`, and a small connector package. diff --git a/python/pyproject.toml b/python/pyproject.toml index 171ce9d..3b2539f 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "rewind-agent" -version = "0.16.0" +version = "0.16.1" description = "Chrome DevTools for AI agents — record, inspect, fork, replay, diff." readme = "README.md" license = "MIT" diff --git a/python/rewind_agent/__init__.py b/python/rewind_agent/__init__.py index 0d0af4f..0d95131 100644 --- a/python/rewind_agent/__init__.py +++ b/python/rewind_agent/__init__.py @@ -15,6 +15,10 @@ def search(query): # Wrap LangGraph / CrewAI for automatic instrumentation graph = rewind_agent.wrap_langgraph(graph) crew = rewind_agent.wrap_crew(crew) + + # One-call connector for custom agents (any HTTP transport, any framework) + with rewind_agent.connector.setup(name="my-agent"): + run_agent_loop() """ from .patch import init, uninit, session, replay, thread @@ -31,6 +35,7 @@ def search(query): ) from .cached_call import cached_llm_call from .explicit import ExplicitClient, RewindReplayDivergenceError +from . import connector from .assertions import Assertions, AssertionResult from .openai_agents import openai_agents_hooks from .pydantic_ai import pydantic_ai_hooks @@ -92,6 +97,8 @@ def search(query): "RewindReplayDivergenceError", # Cached LLM call decorator (Phase 2 / Tier 2) "cached_llm_call", + # One-call connector for any agent (see docs/hdk.md) + "connector", ] @@ -113,4 +120,4 @@ def import_from_langfuse(trace_id: str, **kwargs) -> str: return _import(trace_id, **kwargs) -__version__ = "0.16.0" +__version__ = "0.16.1" diff --git a/python/rewind_agent/connector.py b/python/rewind_agent/connector.py new file mode 100644 index 0000000..ceab305 --- /dev/null +++ b/python/rewind_agent/connector.py @@ -0,0 +1,214 @@ +"""One-call connector for any agent that wants Rewind recording. + +The Tier-1 ergonomic wrapper described in [docs/hdk.md](docs/hdk.md). +Composes :class:`ExplicitClient` + :func:`intercept.install` in the +correct order so callers don't have to remember the dance — and +don't hit the silent-no-op trap where intercept records nothing +because no session is active. + +Usage +----- + +>>> import rewind_agent +>>> with rewind_agent.connector.setup(name="my-agent"): +... run_agent_loop() + +For a custom LLM gateway hostname, pass ``llm_hosts`` (or set +``REWIND_LLM_HOSTS`` in the env): + +>>> with rewind_agent.connector.setup( +... name="my-agent", +... llm_hosts=("llm-gateway.corp.example",), +... ): +... ... + +The yielded value is the underlying :class:`ExplicitClient`, available +for non-HTTP record paths inside the block: + +>>> with rewind_agent.connector.setup(name="my-agent") as client: +... resp = my_grpc_llm.chat(req) # not HTTP — intercept can't see it +... client.record_llm_call( +... request=req, response=resp.dict(), +... model="my-private-llm", duration_ms=duration_ms, +... ) + +This is a *sync* context manager (``@contextmanager``). Async callers +should still use plain ``with``, not ``async with``: the body of the +block can be async — only the setup and teardown are synchronous. + +Environment variables (env > default; kwargs override env) +---------------------------------------------------------- + +* ``REWIND_ENABLED`` — set to ``0`` to make ``setup()`` a no-op with + zero overhead. Yields ``None`` instead of a client; the ``with`` + block runs unmodified. +* ``REWIND_URL`` — Rewind server URL. Default ``http://127.0.0.1:4800``. +* ``REWIND_LLM_HOSTS`` — comma-separated hostnames to treat as LLM + gateways in addition to the strict-by-default provider list. + +Replay-context interaction +-------------------------- + +If ``REWIND_SESSION_ID``, ``REWIND_REPLAY_CONTEXT_ID`` *and* +``REWIND_REPLAY_CONTEXT_TIMELINE_ID`` are all set in the environment +(the runner subprocess pattern documented in docs/runners.md), +``setup()`` skips creating a fresh session and lets +``intercept.install()`` attach to the existing replay context. This +makes the connector safe to drop into runner-driven replay handlers +without phantom sessions. + +All three env vars are required — without ``REWIND_REPLAY_CONTEXT_TIMELINE_ID`` +live cache misses during replay land on an undefined timeline. If only +some are set, the connector treats it as a misconfigured replay and +falls through to the normal session-start path so recording still works. +""" + +from __future__ import annotations + +import logging +import os +from contextlib import contextmanager +from typing import Iterator, Sequence + +from rewind_agent.explicit import ExplicitClient +from rewind_agent.intercept import ( + DefaultPredicates, + install, + is_installed, + uninstall, +) + +logger = logging.getLogger(__name__) + + +class _HostPredicates(DefaultPredicates): + """DefaultPredicates extended with caller-provided LLM gateway hostnames.""" + + def __init__(self, hosts: Sequence[str]) -> None: + super().__init__() + self._hosts = tuple(h for h in (s.strip().lower() for s in hosts) if h) + + def is_llm_call(self, req) -> bool: # type: ignore[override] + netloc = req.url_parts.netloc.lower() + if any(h in netloc for h in self._hosts): + return True + return super().is_llm_call(req) + + +def _resolve_hosts(llm_hosts: Sequence[str] | None) -> tuple[str, ...]: + """Collect extra LLM gateway hostnames from kwarg or ``$REWIND_LLM_HOSTS``. + + Returns an empty tuple when nothing is configured. Callers MUST treat + ``()`` as "use intercept's :class:`DefaultPredicates`" — do not wrap + an empty :class:`_HostPredicates` here, that would silently broaden + matching to no hosts (always-False fallback) which is not the same as + the strict-by-default provider list. + """ + if llm_hosts is not None: + return tuple(llm_hosts) + env = os.environ.get("REWIND_LLM_HOSTS", "") + return tuple(h for h in env.split(",") if h.strip()) if env else () + + +def _enabled(enabled: bool | None) -> bool: + if enabled is not None: + return enabled + return os.environ.get("REWIND_ENABLED", "1") != "0" + + +def _is_replay_dispatch() -> bool: + """Detect runner-subprocess replay env vars. + + When all three are set, intercept.install() will attach to the existing + replay context — we must NOT start a fresh session in that case. + + All three (``REWIND_SESSION_ID``, ``REWIND_REPLAY_CONTEXT_ID``, + ``REWIND_REPLAY_CONTEXT_TIMELINE_ID``) are required for replay mode to + activate. If only some are set, the bootstrap path inside + ``intercept._install`` warns that live cache misses will not have a + defined recording target — which is a half-broken replay. Treat + "partial replay env" as "not replay" so the connector falls through + to the normal session-start path and recording works correctly. + """ + # Truthiness, not `is not None`: an empty-string env var is the same + # as unset for our purposes (`REWIND_SESSION_ID=""` is operator + # misconfiguration, not a valid id). Don't "fix" this to `is not None`. + return bool( + os.environ.get("REWIND_SESSION_ID") + and os.environ.get("REWIND_REPLAY_CONTEXT_ID") + and os.environ.get("REWIND_REPLAY_CONTEXT_TIMELINE_ID") + ) + + +@contextmanager +def setup( + name: str, + *, + base_url: str | None = None, + llm_hosts: Sequence[str] | None = None, + enabled: bool | None = None, + thread_id: str | None = None, + metadata: dict | None = None, +) -> Iterator[ExplicitClient | None]: + """Connect any agent to Rewind for the duration of a ``with`` block. + + Starts a session, installs HTTP intercept (with custom predicates + when ``llm_hosts`` is set), yields the :class:`ExplicitClient` for + use inside the block, and tears both down on exit. + + Parameters + ---------- + name: + Session name shown in ``rewind show`` and the dashboard. + base_url: + Rewind server URL. ``None`` (default) consults + :class:`ExplicitClient`'s resolution: ``$REWIND_URL``, then + ``http://127.0.0.1:4800``. + llm_hosts: + Sequence of hostnames to treat as LLM gateways. ``None`` + (default) reads ``$REWIND_LLM_HOSTS``; empty / unset falls + through to intercept's strict-by-default provider list. + enabled: + ``None`` (default) reads ``$REWIND_ENABLED`` (any value other + than ``"0"`` is on); ``True`` forces on; ``False`` forces off. + When off, ``setup()`` is a true no-op — yields ``None``, no HTTP, + no install. + thread_id, metadata: + Forwarded to :meth:`ExplicitClient.session`. + + Yields + ------ + ExplicitClient | None + The recording client, or ``None`` when disabled. + """ + if not _enabled(enabled): + yield None + return + + # base_url resolution lives in ExplicitClient.__init__ so all callers + # share a single source of truth (kwarg > $REWIND_URL > localhost). + client = ExplicitClient(base_url=base_url) + hosts = _resolve_hosts(llm_hosts) + predicates = _HostPredicates(hosts) if hosts else None + + if _is_replay_dispatch(): + # Runner-driven replay: intercept.install() will attach to the + # existing replay context via env vars. Don't create a phantom + # session. + already_installed = is_installed() + install(predicates=predicates) + try: + yield client + finally: + if not already_installed: + uninstall() + return + + with client.session(name, thread_id=thread_id, metadata=metadata): + already_installed = is_installed() + install(predicates=predicates) + try: + yield client + finally: + if not already_installed: + uninstall() diff --git a/python/rewind_agent/explicit.py b/python/rewind_agent/explicit.py index b8105dc..8c7c14e 100644 --- a/python/rewind_agent/explicit.py +++ b/python/rewind_agent/explicit.py @@ -33,6 +33,7 @@ def get_pods(cluster: str) -> str: import inspect import json import logging +import os import time import urllib.error import urllib.request @@ -99,8 +100,15 @@ def __init__( class ExplicitClient: """Wire-format-agnostic recording client for the Rewind explicit API.""" - def __init__(self, base_url: str = "http://127.0.0.1:4800"): - self.base_url = base_url.rstrip("/") + def __init__(self, base_url: str | None = None): + # Resolution order: explicit kwarg > $REWIND_URL > localhost default. + # Matches intercept._install._bootstrap_replay_context_from_env so a + # process running rewind on a non-default port (multi-tenant deploys, + # local dev where 4800 is taken) sets REWIND_URL once and every + # ExplicitClient — including the lazy singleton in intercept._flow — + # reads it consistently. + resolved = base_url or os.environ.get("REWIND_URL", "http://127.0.0.1:4800") + self.base_url = resolved.rstrip("/") self._enabled = True self._session_cache: dict[str, tuple[str, str, float]] = {} diff --git a/python/rewind_agent/intercept/_install.py b/python/rewind_agent/intercept/_install.py index df29541..c4f5694 100644 --- a/python/rewind_agent/intercept/_install.py +++ b/python/rewind_agent/intercept/_install.py @@ -129,21 +129,29 @@ def install(predicates: Predicates | None = None) -> None: def _bootstrap_replay_context_from_env() -> None: - """Read REWIND_SESSION_ID + REWIND_REPLAY_CONTEXT_ID - (+ optional REWIND_REPLAY_CONTEXT_TIMELINE_ID) and attach. - - The first two are required and must be set together. If either - is missing, this is a silent no-op (operator either explicitly - attached via the SDK or doesn't want bootstrap behavior). - Misconfiguration (one set, other unset) logs a WARN — likely - an env-var typo. - - **Review #154 round 2 fix:** REWIND_REPLAY_CONTEXT_TIMELINE_ID - is also honored. Without it the subprocess-bootstrap path used - to leave `_timeline_id` unset, with the documented consequence - that live cache misses record into the root timeline instead - of the fork. Now it forwards through to attach_replay_context - so the env-var path matches the direct SDK path. + """Read REWIND_SESSION_ID + REWIND_REPLAY_CONTEXT_ID + + REWIND_REPLAY_CONTEXT_TIMELINE_ID and attach. + + All three are required for env-var bootstrap to fire. None set is + a silent no-op (operator explicitly attached via the SDK or isn't + using bootstrap). Any partial subset (one or two of the three) + logs a WARN and skips the attach — partial config means a live + cache miss would land on an undefined recording target, which is + a half-broken replay strictly worse than failing fast. + + **PR #171 review round 2:** previously this required only the + first two and treated the timeline as optional with a separate + WARN log. That was a contract divergence with ``connector.setup``, + which the reviewer flagged: in the partial-replay path, a fresh + session would be started, then ``intercept.install`` would run + this bootstrap on the still-incomplete env, calling + ``attach_replay_context`` and clobbering the fresh session's + contextvars. Tightening to all-three-or-skip aligns the two + consumers and removes the orphaned-session bug. + + The documented runner-subprocess pattern (docs/runners.md) always + sets all three from the dispatch payload, so no in-tree consumer + is affected by the tightening. """ import os @@ -151,7 +159,9 @@ def _bootstrap_replay_context_from_env() -> None: replay_context_id = os.environ.get("REWIND_REPLAY_CONTEXT_ID") timeline_id = os.environ.get("REWIND_REPLAY_CONTEXT_TIMELINE_ID") - if session_id and replay_context_id: + set_count = sum(bool(v) for v in (session_id, replay_context_id, timeline_id)) + + if set_count == 3: try: from rewind_agent.explicit import ExplicitClient @@ -164,23 +174,20 @@ def _bootstrap_replay_context_from_env() -> None: "rewind: attached to replay context %s (session %s, timeline %s) from env", replay_context_id, session_id, - timeline_id or "", + timeline_id, ) - if not timeline_id: - logger.warning( - "rewind: REWIND_REPLAY_CONTEXT_TIMELINE_ID is not set; " - "live cache misses during this replay will not have a " - "defined recording target. Set it from the dispatch payload's " - "replay_context_timeline_id field for correct fork-timeline binding." - ) except Exception as e: # noqa: BLE001 — log + continue, don't break install() logger.warning("rewind: env-var bootstrap failed: %s", e) - elif session_id or replay_context_id: + elif set_count > 0: logger.warning( - "rewind: REWIND_SESSION_ID and REWIND_REPLAY_CONTEXT_ID must be set together; " - "skipping env-var bootstrap (got session=%s ctx=%s)", + "rewind: env-var bootstrap requires REWIND_SESSION_ID + " + "REWIND_REPLAY_CONTEXT_ID + REWIND_REPLAY_CONTEXT_TIMELINE_ID " + "to be set together; skipping (got session=%s ctx=%s timeline=%s). " + "Live cache misses without all three would land on an undefined " + "recording target — fix the dispatch caller to forward all three.", bool(session_id), bool(replay_context_id), + bool(timeline_id), ) _log_install_status() diff --git a/python/tests/test_connector.py b/python/tests/test_connector.py new file mode 100644 index 0000000..099681e --- /dev/null +++ b/python/tests/test_connector.py @@ -0,0 +1,285 @@ +"""Tests for the one-call connector (rewind_agent.connector.setup).""" + +import os +import threading +import unittest +from http.server import HTTPServer, BaseHTTPRequestHandler +from unittest import mock + +import rewind_agent +from rewind_agent.connector import _HostPredicates, _is_replay_dispatch, setup +from rewind_agent.explicit import ( + ExplicitClient, + _replay_context_id, + _session_id, + _timeline_id, +) +from rewind_agent.intercept import is_installed, uninstall + + +class _MockHandler(BaseHTTPRequestHandler): + """Minimal mock — only what the connector needs.""" + + sessions_started: list = [] + sessions_ended: list = [] + + def do_POST(self): # noqa: N802 — stdlib API + import json + + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length)) if length else {} + + if self.path == "/api/sessions/start": + _MockHandler.sessions_started.append(body) + self._respond(201, { + "session_id": f"s-{len(_MockHandler.sessions_started)}", + "root_timeline_id": f"tl-{len(_MockHandler.sessions_started)}", + }) + elif self.path.endswith("/end"): + _MockHandler.sessions_ended.append(self.path) + self._respond(200, {"session_id": self.path.split("/")[3]}) + else: + self._respond(404, {"error": "unhandled"}) + + def _respond(self, status, body): + import json + + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps(body).encode()) + + def log_message(self, *_): # silence + pass + + +class _ConnectorTestBase(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = HTTPServer(("127.0.0.1", 0), _MockHandler) + cls.port = cls.server.server_address[1] + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + cls.base_url = f"http://127.0.0.1:{cls.port}" + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + + def setUp(self): + _MockHandler.sessions_started = [] + _MockHandler.sessions_ended = [] + _session_id.set(None) + _timeline_id.set(None) + _replay_context_id.set(None) + # Make sure intercept is not lingering from a prior test. + if is_installed(): + uninstall() + + +class TestKillSwitch(_ConnectorTestBase): + def test_disabled_via_env_yields_none_and_no_http(self): + with mock.patch.dict(os.environ, {"REWIND_ENABLED": "0"}, clear=False): + with setup(name="off", base_url=self.base_url) as client: + self.assertIsNone(client) + self.assertFalse(is_installed()) + self.assertEqual(_MockHandler.sessions_started, []) + + def test_disabled_via_kwarg(self): + with setup(name="off", base_url=self.base_url, enabled=False) as client: + self.assertIsNone(client) + self.assertEqual(_MockHandler.sessions_started, []) + + +class TestSessionLifecycle(_ConnectorTestBase): + def test_starts_session_and_installs_intercept(self): + with setup(name="my-agent", base_url=self.base_url) as client: + self.assertIsInstance(client, ExplicitClient) + self.assertEqual(_session_id.get(), "s-1") + self.assertEqual(_timeline_id.get(), "tl-1") + self.assertTrue(is_installed()) + + # Cleanup happened. + self.assertIsNone(_session_id.get()) + self.assertFalse(is_installed()) + self.assertEqual(len(_MockHandler.sessions_started), 1) + self.assertEqual(_MockHandler.sessions_started[0]["name"], "my-agent") + self.assertEqual(len(_MockHandler.sessions_ended), 1) + + def test_does_not_uninstall_intercept_if_already_installed(self): + # Simulate: operator already called intercept.install() at startup. + from rewind_agent.intercept import install as intercept_install + intercept_install() + try: + with setup(name="reentrant", base_url=self.base_url): + self.assertTrue(is_installed()) + # We were not the installer, so we must NOT have uninstalled. + self.assertTrue(is_installed()) + finally: + uninstall() + + def test_propagates_thread_id_and_metadata(self): + with setup( + name="threaded", + base_url=self.base_url, + thread_id="conv-42", + metadata={"app": "test"}, + ): + pass + body = _MockHandler.sessions_started[0] + self.assertEqual(body.get("thread_id"), "conv-42") + self.assertEqual(body.get("metadata"), {"app": "test"}) + + +class TestReplayDispatch(_ConnectorTestBase): + _ALL_REPLAY_VARS = ( + "REWIND_SESSION_ID", + "REWIND_REPLAY_CONTEXT_ID", + "REWIND_REPLAY_CONTEXT_TIMELINE_ID", + ) + + def _replay_env(self, **overrides): + """Build a clean env dict with replay vars set unless overridden.""" + base = {k: v for k, v in os.environ.items() if k not in self._ALL_REPLAY_VARS} + base.update({ + "REWIND_SESSION_ID": "s-replay", + "REWIND_REPLAY_CONTEXT_ID": "ctx-replay", + "REWIND_REPLAY_CONTEXT_TIMELINE_ID": "tl-fork", + }) + base.update(overrides) + return {k: v for k, v in base.items() if v is not None} + + def test_replay_env_skips_session_start(self): + env = self._replay_env(REWIND_URL=self.base_url) + with mock.patch.dict(os.environ, env, clear=True): + self.assertTrue(_is_replay_dispatch()) + with setup(name="replay-handler", base_url=self.base_url) as client: + self.assertIsInstance(client, ExplicitClient) + self.assertTrue(is_installed()) + # No phantom session was created on /api/sessions/start. + self.assertEqual(_MockHandler.sessions_started, []) + self.assertEqual(_MockHandler.sessions_ended, []) + + def test_all_three_required_for_replay_mode(self): + # Each replay var, individually unset, breaks replay-mode detection. + # Without all three, intercept._install warns about an undefined + # recording target — better to fall through to normal session start. + for omit in self._ALL_REPLAY_VARS: + with self.subTest(omit=omit): + env = self._replay_env(**{omit: None}) + with mock.patch.dict(os.environ, env, clear=True): + self.assertFalse( + _is_replay_dispatch(), + f"missing {omit} should disable replay mode", + ) + + def test_partial_replay_env_falls_through_to_session_start(self): + # With only two of three replay vars, the connector must start a + # fresh session AND _install._bootstrap_replay_context_from_env + # must NOT clobber the new session's contextvars by attaching to + # the env-supplied (incomplete) replay context. + # + # This is the regression test for the round-2 review finding: + # previously, the bootstrap fired on partial env and the fresh + # session got orphaned because attach_replay_context overwrote + # _session_id with the env-supplied value. The bootstrap is now + # tightened to require all three; this test asserts the fresh + # session's contextvars are live inside the block. + env = self._replay_env(REWIND_REPLAY_CONTEXT_TIMELINE_ID=None) + with mock.patch.dict(os.environ, env, clear=True): + with setup(name="incomplete-replay", base_url=self.base_url): + # _session_id must be the freshly-created session + # ("s-1" from the mock handler), NOT "s-replay" from + # the env. _replay_context_id must remain unset + # because no replay attach happened. + self.assertEqual(_session_id.get(), "s-1") + self.assertEqual(_timeline_id.get(), "tl-1") + self.assertIsNone(_replay_context_id.get()) + self.assertEqual(len(_MockHandler.sessions_started), 1) + self.assertEqual(_MockHandler.sessions_started[0]["name"], "incomplete-replay") + + +def _fake_req(netloc: str): + """Build the minimal duck-typed RewindRequest a Predicates.is_llm_call needs.""" + class _Parts: + pass + + parts = _Parts() + parts.netloc = netloc + req = type("R", (), {})() + req.url_parts = parts + return req + + +class TestHostPredicates(_ConnectorTestBase): + """Behavioral coverage — assert what is_llm_call returns, not on private state.""" + + def _capture_predicates_via_setup(self, **setup_kwargs): + captured = {} + + def fake_install(predicates=None): + captured["predicates"] = predicates + + with mock.patch("rewind_agent.connector.install", side_effect=fake_install), \ + mock.patch("rewind_agent.connector.uninstall"), \ + mock.patch("rewind_agent.connector.is_installed", return_value=False): + with setup(name="capture", base_url=self.base_url, **setup_kwargs): + pass + return captured["predicates"] + + def test_kwarg_hosts_match_via_substring(self): + preds = self._capture_predicates_via_setup( + llm_hosts=("llm-gateway.example.com",), + ) + self.assertIsInstance(preds, _HostPredicates) + # The kwarg host matches — including substring containment. + self.assertTrue(preds.is_llm_call(_fake_req("llm-gateway.example.com"))) + self.assertTrue(preds.is_llm_call(_fake_req("private-llm-gateway.example.com"))) + # Hosts not in kwarg AND not on the strict-by-default provider list don't match. + self.assertFalse(preds.is_llm_call(_fake_req("unrelated.example.com"))) + + def test_env_hosts_parsed_with_whitespace_and_empties_dropped(self): + with mock.patch.dict( + os.environ, + {"REWIND_LLM_HOSTS": "a.example,b.example , ,c.example"}, + clear=False, + ): + preds = self._capture_predicates_via_setup() + self.assertIsInstance(preds, _HostPredicates) + # All three configured hosts match. + self.assertTrue(preds.is_llm_call(_fake_req("a.example"))) + self.assertTrue(preds.is_llm_call(_fake_req("svc.b.example"))) + self.assertTrue(preds.is_llm_call(_fake_req("c.example:8080"))) + # The empty / whitespace-only entries did NOT become a wildcard: + # an unrelated host that doesn't share a substring with any of + # a/b/c.example must still miss. + self.assertFalse(preds.is_llm_call(_fake_req("unrelated.example"))) + + def test_no_hosts_uses_intercept_default_predicates(self): + # Empty REWIND_LLM_HOSTS and no kwarg → install() gets None, which + # tells intercept to apply its strict-by-default DefaultPredicates. + # We do NOT wrap an empty _HostPredicates here (would silently + # broaden matching to "no hosts" — never True for unknown hosts — + # which IS the same in effect, but it's clearer to delegate to + # intercept's default rather than introduce an empty wrapper.) + with mock.patch.dict(os.environ, {"REWIND_LLM_HOSTS": ""}, clear=False): + preds = self._capture_predicates_via_setup() + self.assertIsNone(preds) + + def test_predicate_falls_through_to_default_provider_list(self): + # When custom hosts don't match, the parent DefaultPredicates handles + # known providers like api.openai.com — verify the chain. + preds = _HostPredicates(("custom-gw.example",)) + self.assertTrue(preds.is_llm_call(_fake_req("custom-gw.example"))) + self.assertTrue(preds.is_llm_call(_fake_req("api.openai.com"))) + self.assertFalse(preds.is_llm_call(_fake_req("unrelated.com"))) + + +class TestPublicExport(unittest.TestCase): + def test_module_attribute(self): + self.assertTrue(hasattr(rewind_agent, "connector")) + self.assertTrue(callable(rewind_agent.connector.setup)) + + +if __name__ == "__main__": + unittest.main() diff --git a/python/tests/test_explicit.py b/python/tests/test_explicit.py index 78fadda..ac7cb9d 100644 --- a/python/tests/test_explicit.py +++ b/python/tests/test_explicit.py @@ -504,5 +504,39 @@ def test_ensure_session_then_record(self): self.assertEqual(len(MockRewindHandler.recorded_steps), 1) +class TestExplicitClientBaseUrlResolution(unittest.TestCase): + """Cover the __init__ resolution order: kwarg > $REWIND_URL > default.""" + + def test_default_base_url(self): + import os + from unittest import mock + + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("REWIND_URL", None) + client = ExplicitClient() + self.assertEqual(client.base_url, "http://127.0.0.1:4800") + + def test_env_var_used_when_kwarg_absent(self): + import os + from unittest import mock + + with mock.patch.dict( + os.environ, {"REWIND_URL": "http://rewind.test:9999/"}, clear=False + ): + client = ExplicitClient() + # Trailing slash gets stripped to match the existing convention. + self.assertEqual(client.base_url, "http://rewind.test:9999") + + def test_kwarg_wins_over_env_var(self): + import os + from unittest import mock + + with mock.patch.dict( + os.environ, {"REWIND_URL": "http://rewind.from-env:9999"}, clear=False + ): + client = ExplicitClient(base_url="http://rewind.from-kwarg:7777") + self.assertEqual(client.base_url, "http://rewind.from-kwarg:7777") + + if __name__ == "__main__": unittest.main() diff --git a/python/tests/test_intercept_install.py b/python/tests/test_intercept_install.py index 9c75642..1277214 100644 --- a/python/tests/test_intercept_install.py +++ b/python/tests/test_intercept_install.py @@ -172,5 +172,105 @@ def upstream_handler(request: httpx.Request) -> httpx.Response: _savings.reset() +class TestEnvVarBootstrapContract(unittest.TestCase): + """Cover the all-three-or-skip contract of _bootstrap_replay_context_from_env. + + Previously the bootstrap required only REWIND_SESSION_ID + + REWIND_REPLAY_CONTEXT_ID and treated the timeline as optional. That + contract diverged from connector.setup() and caused a clobber bug + in the connector's partial-replay fall-through path. PR #171 round 2 + tightened to require all three; partial subsets now skip the attach + with a single WARN. + """ + + def setUp(self) -> None: + uninstall() + # Reset contextvars so we observe attach behavior cleanly. + from rewind_agent.explicit import ( + _replay_context_id, + _session_id, + _timeline_id, + ) + _session_id.set(None) + _timeline_id.set(None) + _replay_context_id.set(None) + + def tearDown(self) -> None: + uninstall() + + def _run_bootstrap_with_env(self, env: dict) -> tuple: + """Run install() under env and return (session_id, replay_ctx_id, timeline_id) contextvars after.""" + import os + from unittest import mock + + from rewind_agent.explicit import ( + _replay_context_id, + _session_id, + _timeline_id, + ) + + # Strip any inherited rewind-related vars so the test env is clean. + clean = {k: v for k, v in os.environ.items() if not k.startswith("REWIND_")} + clean.update(env) + with mock.patch.dict(os.environ, clean, clear=True): + install() + return ( + _session_id.get(), + _replay_context_id.get(), + _timeline_id.get(), + ) + + def test_all_three_set_attaches(self) -> None: + sid, ctx, tid = self._run_bootstrap_with_env({ + "REWIND_SESSION_ID": "s-env", + "REWIND_REPLAY_CONTEXT_ID": "ctx-env", + "REWIND_REPLAY_CONTEXT_TIMELINE_ID": "tl-env", + }) + self.assertEqual(sid, "s-env") + self.assertEqual(ctx, "ctx-env") + self.assertEqual(tid, "tl-env") + + def test_none_set_is_silent_noop(self) -> None: + sid, ctx, tid = self._run_bootstrap_with_env({}) + self.assertIsNone(sid) + self.assertIsNone(ctx) + self.assertIsNone(tid) + + def test_partial_subsets_skip_attach(self) -> None: + """Each partial subset (1 or 2 of 3) must NOT attach. + + This is the contract the connector relies on: when partial + replay env is detected, the bootstrap leaves contextvars alone + so the connector's freshly-started session stays in scope. + """ + all_vars = { + "REWIND_SESSION_ID": "s-env", + "REWIND_REPLAY_CONTEXT_ID": "ctx-env", + "REWIND_REPLAY_CONTEXT_TIMELINE_ID": "tl-env", + } + keys = list(all_vars.keys()) + partials: list[dict] = [] + # All single-var subsets. + for k in keys: + partials.append({k: all_vars[k]}) + # All two-var subsets. + for i, k1 in enumerate(keys): + for k2 in keys[i + 1 :]: + partials.append({k1: all_vars[k1], k2: all_vars[k2]}) + + for env in partials: + with self.subTest(env_keys=sorted(env.keys())): + sid, ctx, tid = self._run_bootstrap_with_env(env) + self.assertIsNone( + sid, f"REWIND_SESSION_ID must not be attached with {env}" + ) + self.assertIsNone( + ctx, f"_replay_context_id must not be set with {env}" + ) + self.assertIsNone( + tid, f"_timeline_id must not be set with {env}" + ) + + if __name__ == "__main__": unittest.main() diff --git a/python/tests/test_runner.py b/python/tests/test_runner.py index 51614a2..d74d263 100644 --- a/python/tests/test_runner.py +++ b/python/tests/test_runner.py @@ -242,9 +242,19 @@ def test_attach_replay_context_sets_contextvars() -> None: assert _replay_context_id.get() == "ctx-attach" -def test_install_bootstraps_from_env(monkeypatch: pytest.MonkeyPatch) -> None: - """``intercept.install()`` reads REWIND_SESSION_ID + - REWIND_REPLAY_CONTEXT_ID and attaches before patching. +def test_install_bootstraps_only_when_all_three_env_vars_set( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """PR #171 round 2: bootstrap requires all three env vars + (REWIND_SESSION_ID + REWIND_REPLAY_CONTEXT_ID + + REWIND_REPLAY_CONTEXT_TIMELINE_ID). + + Previously, the bootstrap fired with only the first two and treated + the timeline as optional with a WARN. That contract diverged from + connector.setup() and caused a clobber bug — see + test_intercept_install.TestEnvVarBootstrapContract for full coverage. + Here we just sanity-check the subprocess path: missing timeline + means no attach. """ from rewind_agent.explicit import _replay_context_id, _session_id, _timeline_id from rewind_agent.intercept import _install @@ -261,8 +271,9 @@ def test_install_bootstraps_from_env(monkeypatch: pytest.MonkeyPatch) -> None: try: _install._bootstrap_replay_context_from_env() - assert _session_id.get() == "boot-sess" - assert _replay_context_id.get() == "boot-ctx" + # Without REWIND_REPLAY_CONTEXT_TIMELINE_ID, no attach happens. + assert _session_id.get() is None + assert _replay_context_id.get() is None assert _timeline_id.get() is None finally: _session_id.set(None) @@ -304,19 +315,29 @@ def test_install_bootstraps_with_timeline_id_from_env( def test_install_partial_env_logs_warning_and_skips( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - from rewind_agent.explicit import _replay_context_id, _session_id + """Partial env-var subset (any 1 or 2 of 3) → WARN + skip attach. + + The new contract requires all three together; a partial subset is + operator misconfiguration that should fail loudly rather than + half-attach with a warning (which used to leave live cache misses + on an undefined timeline). + """ + from rewind_agent.explicit import _replay_context_id, _session_id, _timeline_id from rewind_agent.intercept import _install monkeypatch.setenv("REWIND_SESSION_ID", "only-session") monkeypatch.delenv("REWIND_REPLAY_CONTEXT_ID", raising=False) + monkeypatch.delenv("REWIND_REPLAY_CONTEXT_TIMELINE_ID", raising=False) _session_id.set(None) _replay_context_id.set(None) + _timeline_id.set(None) with caplog.at_level("WARNING"): _install._bootstrap_replay_context_from_env() assert _session_id.get() is None assert _replay_context_id.get() is None + assert _timeline_id.get() is None assert any( - "must be set together" in r.message for r in caplog.records + "set together" in r.message for r in caplog.records ), f"records={[r.message for r in caplog.records]}"