RFC: AIGP-Light — Minimal Detect & Respond Client

PRIVATE AND PROPRIETARY. Owned by Kanjani AI Research & Causum. See NOTICE.md.

Status: Draft RFC Author: Kanjani AI Research & Causum Date: July 2026 Target: sdks/python/aigp-light/ Companion: Position Paper — “AIGP-Light: Graduated Governance Through Minimal Detect & Respond”

1. Purpose

This RFC specifies the architecture, interfaces, and scope boundaries for AIGP-Light — a minimal, local-first governance client that provides detect & respond capability at the model call boundary with zero external dependencies.

AIGP-Light emits traces in RFC-010 §5.3 format and is forward-compatible with the full AIGP protocol stack.

2. Scope Boundaries

2.1 In Scope

Concern	Implementation
Model call interception	Wrap any `callable → response` boundary
Behavioral detection	Windowed statistical signals (latency, refusal, error, cost, token volume)
Policy enforcement	Local YAML-driven response actions (LOG, ALERT, RATE_LIMIT, FALLBACK, BLOCK, CIRCUIT_BREAK)
Trace emission	RFC-010 §5.3 spans (stage 9 + stage 26)
Local storage	JSONL file, stdout, or callback
Framework adapters	OpenAI, Anthropic, LangChain (thin wrappers over core)

2.2 Out of Scope (deferred to full AIGP)

Concern	Why Excluded
External governance service	Zero-network-dependency contract
VLT / delegation tokens	Multi-agent concern (Level 3+)
Consent providers	Requires infrastructure
Regulatory contexts	Jurisdictional awareness requires full client
Streaming governance	Token-by-token requires buffer management
Anticipation engine	Predictive policy requires model state
Evidence graph / Merkle DAG	Cryptographic linking requires infrastructure
Dialect subscription	Active Receiver consumption requires service
Multi-agent orchestration	Single call boundary only

3. Architecture

flowchart TD
    subgraph APP["Application"]
        A["gov = AigpLight.from_yaml('aigp-light.yaml')<br/>response = gov.call(model.chat, messages=msgs)"]
    end

    APP --> LIGHT

    subgraph LIGHT["aigp_light"]
        direction LR
        subgraph INT["Interceptor"]
            B1["• pre_call"]
            B2["• post_call"]
            B3["• on_error"]
        end
        subgraph POL["PolicyEngine"]
            C1["• evaluate()"]
            C2["• respond()"]
            C3["• state"]
        end
        subgraph TRC["TraceEmitter"]
            D1["• emit()"]
            D2["• flush()"]
            D3["• export()"]
        end
        INT -->|signals| POL
        POL -->|actions| TRC
        INT --> SS["SignalStore<br/>(windowed)"]
        POL --> CFG["Policy<br/>(YAML)"]
    end

    style APP fill:#34495e,color:#ecf0f1
    style LIGHT fill:#2c3e50,color:#ecf0f1

### 3.1 Module Layout

sdks/python/aigp-light/ ├── pyproject.toml ├── README.md ├── aigp_light/ │ ├── init.py # Public API: AigpLight, PolicyEngine, TraceEmitter │ ├── core.py # AigpLight orchestrator class │ ├── interceptor.py # Call interception + signal extraction │ ├── policy.py # PolicyEngine — threshold evaluation + response │ ├── signals.py # SignalStore — windowed metric accumulation │ ├── responses.py # Response action implementations │ ├── trace.py # TraceEmitter — RFC-010 §5.3 span builder │ ├── config.py # YAML policy loader + validation │ └── adapters/ │ ├── init.py │ ├── openai.py # GovernedOpenAI (light) │ ├── anthropic.py # GovernedAnthropic (light) │ └── langchain.py # GovernedLangChain (light) └── tests/ ├── test_core.py ├── test_policy.py ├── test_signals.py ├── test_trace.py └── test_adapters.py

---

## 4. Core Interface

### 4.1 `AigpLight` — Primary Entry Point

```python
class AigpLight:
    """Minimal AIGP governance — detect & respond at model call boundary."""

    @classmethod
    def from_yaml(cls, path: str) -> "AigpLight":
        """Load policy from YAML configuration file."""

    @classmethod
    def from_dict(cls, config: dict) -> "AigpLight":
        """Load policy from dictionary (programmatic configuration)."""

    def call(self, fn: Callable, *args, **kwargs) -> Any:
        """Govern a synchronous model call.

        Intercepts the call, evaluates policy, executes response
        actions if thresholds breached, emits trace span.
        """

    async def acall(self, fn: Callable, *args, **kwargs) -> Any:
        """Govern an async model call."""

    def wrap(self, fn: Callable) -> Callable:
        """Return a governed wrapper around fn (decorator pattern)."""

    @property
    def metrics(self) -> dict:
        """Current windowed signal state (for inspection/debugging)."""

    @property
    def traces(self) -> list[dict]:
        """Buffered traces not yet flushed."""

    def flush(self) -> None:
        """Force flush buffered traces to configured output."""

    def reset(self) -> None:
        """Reset all windowed signal state (for testing)."""

4.2 Usage Examples

Minimal (5 lines):

from aigp_light import AigpLight

gov = AigpLight.from_yaml("aigp-light.yaml")
response = gov.call(client.chat.completions.create,
                    model="gpt-4o", messages=messages)

Decorator pattern:

from aigp_light import AigpLight

gov = AigpLight.from_yaml("aigp-light.yaml")

@gov.wrap
def chat(model, messages, **kwargs):
    return client.chat.completions.create(model=model, messages=messages, **kwargs)

response = chat("gpt-4o", messages)

Async:

response = await gov.acall(client.chat.completions.create,
                           model="gpt-4o", messages=messages)

Framework adapter (OpenAI):

from aigp_light.adapters.openai import LightOpenAI

governed = LightOpenAI(client=OpenAI(), policy="aigp-light.yaml")
response = governed.chat("gpt-4o", messages)

5. Policy Engine

5.1 `PolicyEngine` Interface

@dataclass
class PolicyResult:
    """Result of policy evaluation against current signals."""
    triggered: bool
    breaches: list[Breach]
    actions: list[ResponseAction]

@dataclass
class Breach:
    signal: str           # e.g. "latency.p95"
    threshold: float
    actual: float
    window_seconds: int

class PolicyEngine:
    """Evaluates detection signals against configured thresholds."""

    def __init__(self, config: dict):
        """Initialize with parsed YAML config."""

    def evaluate(self, signals: SignalSnapshot) -> PolicyResult:
        """Evaluate current signal snapshot against all thresholds.

        Returns PolicyResult with any breaches and their mapped actions.
        Pure function — no side effects.
        """

    def respond(self, result: PolicyResult) -> ResponseAction:
        """Determine the highest-priority response action.

        Priority: CIRCUIT_BREAK > BLOCK > FALLBACK > RATE_LIMIT > ALERT > LOG
        """

    @property
    def circuit_open(self) -> bool:
        """True if circuit breaker is currently active."""

    @property
    def rate_limited(self) -> bool:
        """True if rate limiter is currently throttling."""

5.2 Policy Configuration Schema

# aigp-light.yaml — full schema
version: "1.0"

# Identity (optional, included in traces)
app_id: "my-app"
model_id: "gpt-4o"

# Detection thresholds
detection:
  latency:
    p95_threshold_ms: 3000       # p95 latency over window
    p99_threshold_ms: 8000       # p99 latency over window
    window_seconds: 60

  refusal_rate:
    threshold: 0.15              # fraction of calls classified as refusal
    window_seconds: 300
    # Refusal detection: response contains refusal patterns
    patterns:
      - "I cannot"
      - "I'm unable"
      - "I apologize, but"
      - "As an AI"

  error_rate:
    threshold: 0.05              # fraction of calls that raise exceptions
    window_seconds: 60

  token_explosion:
    multiplier: 3.0              # output_tokens > multiplier × rolling_average
    baseline_window_seconds: 3600

  cost:
    max_per_minute_usd: 1.00
    # Pricing (per 1M tokens, configurable per model)
    pricing:
      input_per_million: 2.50
      output_per_million: 10.00

# Response actions mapped to threshold breaches
response:
  latency.p95_threshold_ms:
    actions: [LOG, ALERT]
  latency.p99_threshold_ms:
    actions: [LOG, ALERT, RATE_LIMIT]
  refusal_rate.threshold:
    actions: [LOG, ALERT]
  error_rate.threshold:
    actions: [LOG, CIRCUIT_BREAK]
    circuit_break:
      cooldown_seconds: 30
  token_explosion.multiplier:
    actions: [LOG, FALLBACK]
    fallback:
      model_id: "gpt-4o-mini"   # fallback model
  cost.max_per_minute_usd:
    actions: [RATE_LIMIT, ALERT]
    rate_limit:
      max_calls_per_second: 1

# Trace output configuration
trace:
  output: file                   # file | stdout | callback
  path: ./aigp-traces/
  format: jsonl
  buffer_size: 10                # flush every N spans
  include_detection: true        # include signal snapshot in span attributes

# Alert configuration (optional)
alerts:
  webhook_url: null              # POST JSON to this URL
  callback: null                 # Python callable path (e.g. "myapp.alerts.notify")

6. Signal Store

6.1 `SignalStore` Interface

@dataclass
class SignalSnapshot:
    """Point-in-time view of all detection signals."""
    latency_p95_ms: float
    latency_p99_ms: float
    refusal_rate: float
    error_rate: float
    token_output_avg: float
    token_output_last: int
    cost_per_minute_usd: float
    total_calls: int
    window_calls: int
    timestamp: float

class SignalStore:
    """Windowed metric accumulator for behavioral detection.

    Maintains time-bucketed counters for each detection signal.
    Thread-safe for concurrent access.
    """

    def __init__(self, config: dict):
        """Initialize with detection config (window sizes, etc.)."""

    def record(self, observation: CallObservation) -> None:
        """Record a completed model call observation.

        Updates all relevant signal windows.
        """

    def snapshot(self) -> SignalSnapshot:
        """Return current signal state across all windows.

        Calculates percentiles, rates, and averages from
        the current window contents.
        """

    def reset(self) -> None:
        """Clear all accumulated signals (testing only)."""

@dataclass
class CallObservation:
    """Raw observation from a single model call."""
    timestamp: float
    duration_ms: int
    input_tokens: int
    output_tokens: int
    success: bool
    refusal: bool
    error: Exception | None = None
    model_id: str = ""

6.2 Windowing Strategy

Signals use tumbling windows with configurable duration:

gantt
    title Signal Windowing Strategy (Tumbling Windows)
    dateFormat X
    axisFormat %s

    section Window 1 (60s)
        obs obs obs obs obs — p95 calculated : 0, 60

    section Window 2 (60s)
        obs obs obs — p95 calculated : 60, 120

    section Window 3 (60s)
        obs ... : 120, 180

Each signal type has its own window duration (configured in YAML)
Observations older than the window are evicted on read
Percentiles use a sorted buffer (not reservoir sampling — call volume is bounded)
Thread safety via threading.Lock on each signal buffer

7. Response Actions

7.1 Action Implementations

class ResponseAction(ABC):
    """Base class for response actions."""

    @abstractmethod
    def execute(self, context: ActionContext) -> ActionResult:
        """Execute the response action. Returns result."""

@dataclass
class ActionContext:
    """Context available to response actions."""
    breach: Breach
    call_args: tuple
    call_kwargs: dict
    policy_config: dict
    signal_snapshot: SignalSnapshot

@dataclass
class ActionResult:
    """Result of executing a response action."""
    action: str                  # LOG, ALERT, RATE_LIMIT, etc.
    applied: bool                # Whether the action changed behavior
    details: dict                # Action-specific details


class LogAction(ResponseAction):
    """Emit structured log event. Never blocks the call."""

class AlertAction(ResponseAction):
    """Send alert notification (webhook or callback). Non-blocking."""

class RateLimitAction(ResponseAction):
    """Enforce call rate limit. May delay the call (sleep) or reject."""

class FallbackAction(ResponseAction):
    """Substitute model/parameters. Modifies the call target."""

class BlockAction(ResponseAction):
    """Reject the call entirely. Raises GovernanceBlockedError."""

class CircuitBreakAction(ResponseAction):
    """Open circuit for cooldown period. All calls rejected until reset."""

7.2 Action Priority and Composition

When multiple thresholds breach simultaneously, actions compose with priority ordering:

CIRCUIT_BREAK (6) > BLOCK (5) > FALLBACK (4) > RATE_LIMIT (3) > ALERT (2) > LOG (1)

Rules:

Mutually exclusive at top: If CIRCUIT_BREAK fires, BLOCK/FALLBACK/RATE_LIMIT are moot
Additive at bottom: LOG and ALERT always execute regardless of higher-priority actions
FALLBACK replaces, doesn’t stack: Only one fallback target applies

Example: If both error_rate (→ CIRCUIT_BREAK) and cost (→ RATE_LIMIT) breach simultaneously:

CIRCUIT_BREAK wins (priority 6 > 3)
LOG fires for both breaches (additive)
ALERT fires for both breaches (additive)
RATE_LIMIT does not execute (circuit is open)

8. Trace Emitter

8.1 `TraceEmitter` Interface

class TraceEmitter:
    """Emits RFC-010 §5.3 compatible trace spans.

    Simplified version of aigp_client.TraceBuilder — uses only
    stage 9 (runtime_invocation) and stage 26 (circuit_breaker_evaluation).
    """

    def __init__(self, config: dict):
        """Initialize with trace output config."""

    def emit(self, span: TraceSpan) -> None:
        """Buffer a trace span for output."""

    def flush(self) -> None:
        """Write buffered spans to configured output (file/stdout/callback)."""

    def export(self) -> list[dict]:
        """Return all spans as dicts (for programmatic access)."""

@dataclass
class TraceSpan:
    """A single RFC-010 §5.3 compatible span."""
    trace_id: str
    stage: int                   # 9 or 26
    stage_name: str
    start_time: str              # ISO 8601
    end_time: str                # ISO 8601
    duration_ms: int
    status: str                  # OK | ERROR | BLOCKED | CIRCUIT_OPEN
    attributes: dict             # model_id, tokens, detection signals, actions

8.2 Trace Output Format

Each emitted trace is a complete JSON object (one per line in JSONL mode):

{
  "trace_id": "trc-light-a1b2c3d4e5f6",
  "app_id": "my-app",
  "protocol": "aigp-light/1.0",
  "spans": [
    {
      "stage": 9,
      "stage_name": "runtime_invocation",
      "start_time": "2026-07-11T11:00:00.123Z",
      "end_time": "2026-07-11T11:00:01.456Z",
      "duration_ms": 1333,
      "status": "OK",
      "attributes": {
        "model_id": "gpt-4o",
        "input_tokens": 1200,
        "output_tokens": 450,
        "detection": {
          "latency_p95_ms": 1280,
          "refusal_rate": 0.02,
          "error_rate": 0.00,
          "cost_rate_usd_min": 0.34
        },
        "policy": {
          "triggered": false,
          "actions": []
        }
      }
    }
  ],
  "summary": {
    "total_duration_ms": 1333,
    "stages_traversed": 1,
    "slowest_stage": 9,
    "error_stages": [],
    "policy_actions": []
  }
}

When a policy action fires, an additional stage 26 span appears:

{
  "stage": 26,
  "stage_name": "circuit_breaker_evaluation",
  "start_time": "2026-07-11T11:00:01.456Z",
  "end_time": "2026-07-11T11:00:01.457Z",
  "duration_ms": 1,
  "status": "TRIGGERED",
  "attributes": {
    "breaches": [
      {"signal": "error_rate.threshold", "threshold": 0.05, "actual": 0.12}
    ],
    "actions_applied": ["LOG", "CIRCUIT_BREAK"],
    "circuit_cooldown_seconds": 30
  }
}

8.3 Forward Compatibility

The protocol: "aigp-light/1.0" field distinguishes Light traces from full AIGP traces. Full AIGP consumers:

Recognize the format (RFC-010 §5.3 spans — identical structure)
Can ingest Light traces directly into the evidence stream
Understand that stages other than 9/26 are absent (Light doesn’t traverse them)
Can retroactively link Light traces to a governance service via trace_id

When upgrading from AIGP-Light to full AIGP:

Historical Light traces are importable via trace_id + timestamp
The trust function accrues evidence from both Light and full traces
No data transformation required

9. Interceptor

9.1 Call Lifecycle

class Interceptor:
    """Wraps model calls with detection + response + trace."""

    def __init__(self, policy: PolicyEngine, signals: SignalStore,
                 trace: TraceEmitter, config: dict):
        pass

    def govern(self, fn: Callable, *args, **kwargs) -> Any:
        """Full governed call lifecycle:

        1. PRE-CALL: Check circuit state
           - If circuit open → raise CircuitOpenError
           - If rate limited → sleep or reject

        2. EXECUTE: Call the wrapped function
           - Capture timing, tokens, success/failure
           - Detect refusal patterns in response

        3. POST-CALL: Record + Evaluate + Respond
           - Record observation in SignalStore
           - Evaluate signals against policy
           - Execute response actions (if triggered)
           - Emit trace span

        4. RETURN: response (or FALLBACK response, or raise)
        """

9.2 Signal Extraction

The interceptor extracts signals from model responses using a pluggable extraction strategy:

class SignalExtractor(Protocol):
    """Extract observation signals from a model response."""

    def extract(self, response: Any, duration_ms: int,
                error: Exception | None) -> CallObservation:
        ...

class OpenAIExtractor(SignalExtractor):
    """Extract from OpenAI ChatCompletion response."""

    def extract(self, response, duration_ms, error):
        if error:
            return CallObservation(
                timestamp=time.time(), duration_ms=duration_ms,
                input_tokens=0, output_tokens=0,
                success=False, refusal=False, error=error
            )
        return CallObservation(
            timestamp=time.time(),
            duration_ms=duration_ms,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            success=True,
            refusal=self._detect_refusal(response),
            model_id=response.model,
        )

    def _detect_refusal(self, response) -> bool:
        """Check response content against refusal patterns."""
        content = response.choices[0].message.content or ""
        return any(p in content for p in self._patterns)

10. Framework Adapters

10.1 Adapter Pattern

Each adapter is a thin wrapper that provides framework-specific ergonomics over the core AigpLight.call():

class LightOpenAI:
    """AIGP-Light governed OpenAI client.

    Drop-in wrapper — same interface as GovernedOpenAI from
    aigp-agent-core, but local-only (no external service).
    """

    def __init__(self, client, policy: str | dict):
        self._client = client
        self._gov = AigpLight.from_yaml(policy) if isinstance(policy, str) else AigpLight.from_dict(policy)

    def chat(self, model: str, messages: list, **kwargs) -> Any:
        """Governed chat completion."""
        return self._gov.call(
            self._client.chat.completions.create,
            model=model, messages=messages, **kwargs
        )

    async def achat(self, model: str, messages: list, **kwargs) -> Any:
        """Async governed chat completion."""
        return await self._gov.acall(
            self._client.chat.completions.create,
            model=model, messages=messages, **kwargs
        )

10.2 Adapter Interface Compatibility

Light adapters mirror the same call interface pattern, making codebases consistent whether using D&R alone or with governance:

Method	Light Adapter	Full AIGP Adapter
`chat()`	Local D&R policy	CHECK → execute → RECORD → TRACE
`achat()`	Local D&R policy	Same, async
Constructor	`policy: str` (YAML path)	`gov_url, app_id, hmac_secret`

The call site (governed.chat(...)) is identical in both cases — emitter compatible by design.

11. Error Handling

11.1 Exception Hierarchy

class AigpLightError(Exception):
    """Base exception for all AIGP-Light errors."""

class GovernanceBlockedError(AigpLightError):
    """Call blocked by policy (BLOCK action)."""
    breach: Breach

class CircuitOpenError(AigpLightError):
    """Call rejected — circuit breaker is open."""
    cooldown_remaining_seconds: float
    opened_at: float
    reason: Breach

class RateLimitedError(AigpLightError):
    """Call rejected — rate limit exceeded (when non-blocking mode)."""
    retry_after_seconds: float

class PolicyConfigError(AigpLightError):
    """Invalid policy YAML configuration."""
    path: str
    details: str

11.2 Failure Modes

Scenario	Behavior
Policy YAML missing/invalid	Raise `PolicyConfigError` at construction time
Model call raises exception	Record as error observation → evaluate policy → emit trace with ERROR status
Trace write fails (disk full)	Log warning, do NOT block the model call
Alert webhook fails	Log warning, do NOT block the model call
Signal store overflow	Evict oldest observations (bounded memory)

Principle: AIGP-Light governance failures must never cause application failures. The model call always succeeds if the model itself succeeds — governance is additive safety, not a new failure mode.

12. Testing Strategy

12.1 Unit Tests

# test_policy.py — Policy engine in isolation
def test_no_breach_when_under_threshold():
    engine = PolicyEngine(config_with_latency_3000ms)
    result = engine.evaluate(snapshot_with_latency_2000ms)
    assert not result.triggered

def test_breach_triggers_correct_actions():
    engine = PolicyEngine(config_with_error_rate_005)
    result = engine.evaluate(snapshot_with_error_rate_012)
    assert result.triggered
    assert "CIRCUIT_BREAK" in [a.action for a in result.actions]

# test_signals.py — Windowed accumulation
def test_window_eviction():
    store = SignalStore(window_60s)
    store.record(observation_at_t_minus_120s)
    store.record(observation_at_t_minus_30s)
    snap = store.snapshot()
    assert snap.window_calls == 1  # old observation evicted

# test_trace.py — RFC-010 §5.3 compliance
def test_trace_span_format():
    emitter = TraceEmitter(config_stdout)
    emitter.emit(span)
    output = emitter.export()
    assert output[0]["stage"] == 9
    assert "start_time" in output[0]
    assert output[0]["start_time"].endswith("Z")

12.2 Integration Tests

# test_core.py — Full lifecycle
def test_governed_call_success():
    gov = AigpLight.from_dict(permissive_policy)
    result = gov.call(mock_model_success)
    assert result == mock_response
    assert len(gov.traces) == 1
    assert gov.traces[0]["spans"][0]["status"] == "OK"

def test_circuit_break_on_errors():
    gov = AigpLight.from_dict(strict_error_policy)
    for _ in range(10):
        with suppress(Exception):
            gov.call(mock_model_error)
    with pytest.raises(CircuitOpenError):
        gov.call(mock_model_success)

def test_fallback_on_token_explosion():
    gov = AigpLight.from_dict(token_explosion_policy_with_fallback)
    # Fill baseline window
    for _ in range(100):
        gov.call(mock_model_normal_tokens)
    # Trigger explosion
    result = gov.call(mock_model_huge_tokens)
    # Should have called fallback model instead
    assert result.model == "gpt-4o-mini"

13. AIGP Governance Integration (Optional)

AIGP-Light is a standalone product. It does not require AIGP governance.

If an organization also uses AIGP governance, traces can be forwarded via the receiver plugin — no application code change required:

# Standalone D&R (default):
from aigp_light import AigpLight
gov = AigpLight.from_yaml("aigp-light.yaml")
response = gov.call(model.chat, messages=msgs)

# Same code + AIGP governance receiver (add one parameter):
gov = AigpLight.from_yaml("aigp-light.yaml", receiver="https://gov.example.com")
response = gov.call(model.chat, messages=msgs)
# → D&R runs locally (detect, respond, trace)
# → Traces ALSO forwarded to AIGP governance service

The receiver= parameter registers the AigpReceiver plugin. The emitter (gov.call()) never changes. D&R still runs locally regardless of whether the governance service is reachable.

13.1 Trace Compatibility

AIGP-Light traces use RFC-010 §5.3 format — the same format used by AIGP governance. This is a shared schema, not an upgrade dependency. Both products read the same format independently.

14. Dependencies

14.1 Runtime Dependencies

aigp-light:
  python: ">=3.10"
  dependencies: []    # ZERO external dependencies

AIGP-Light has zero runtime dependencies beyond the Python standard library. This is a hard requirement:

YAML parsing: yaml is stdlib (via pyyaml — the ONE optional dependency, with fallback to JSON config)
HTTP for webhooks: urllib.request (stdlib)
JSON: stdlib
Threading: stdlib
Time: stdlib

14.2 Optional Dependencies

[project.optional-dependencies]
yaml = ["pyyaml>=6.0"]          # YAML config support (vs JSON-only)
openai = ["openai>=1.0"]        # OpenAI adapter
anthropic = ["anthropic>=0.20"] # Anthropic adapter
langchain = ["langchain-core>=0.2"]  # LangChain adapter

14.3 Development Dependencies

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "pytest-cov>=5.0",
    "ruff>=0.4",
    "mypy>=1.10",
]

15. Non-Functional Requirements

Requirement	Target	Rationale
Policy evaluation latency	< 100μs	Must not measurably impact model call latency
Memory per signal window	< 1MB per 10K observations	Bounded memory for long-running services
Trace write latency	< 1ms (buffered)	Async flush, never blocks caller
Thread safety	Full	Must work in multi-threaded servers
Async support	Native	Must work in asyncio event loops
Cold start	< 10ms	Policy load + validation at construction
Package size	< 50KB	Minimal install footprint

16. Security Considerations

Concern	Mitigation
Policy file tampering	Document that policy file integrity is the deployer’s responsibility. Optionally support SHA-256 checksum validation.
Trace data sensitivity	Traces contain model_id, token counts, latency — no prompt/response content by default. Content inclusion is opt-in.
Webhook secret	Alert webhooks support HMAC signing for receiver authentication.
Denial of service via policy	Malformed policy cannot crash the process — validation at load time. Invalid policy → `PolicyConfigError`, not silent failure.

17. Open Questions

#	Question	Options	Recommendation
1	Should AIGP-Light detect refusals via regex or embedding?	Regex (zero-dep) vs embedding (requires model)	Regex — zero dependencies is a hard constraint
2	Should traces include prompt/response content?	Never / opt-in / summarized	Opt-in with explicit `include_content: true` in config
3	Should AIGP-Light support custom signals?	Fixed set only / pluggable	Pluggable via `SignalExtractor` protocol — but ship with fixed set
4	Where does `pyyaml` fall — required or optional?	Required / optional with JSON fallback	Optional with JSON fallback. Zero-dep means zero-dep.
5	Should circuit breaker state persist across restarts?	In-memory only / file-backed	In-memory only. Stateless restart is safer.

18. Implementation Plan

Phase	Deliverable	Effort
Phase 1	`core.py`, `policy.py`, `signals.py`, `trace.py` — core lifecycle	2 days
Phase 2	`responses.py`, `interceptor.py`, `config.py` — full engine	1 day
Phase 3	`adapters/openai.py` — first adapter	0.5 day
Phase 4	Tests (unit + integration)	1 day
Phase 5	`README.md`, `pyproject.toml`, packaging	0.5 day
Phase 6	Additional adapters (Anthropic, LangChain)	1 day

Total: ~6 days to first release.

Appendix A: Trace Format Cross-Reference

Field	AIGP-Light	Full AIGP (TraceBuilder)	Compatible?
`trace_id`	`trc-light-{hex12}`	`trc-{hex12}`	✓ (prefix differs, format same)
`spans[].stage`	9, 26 only	1–26	✓ (subset)
`spans[].stage_name`	From STAGE_NAMES registry	Same registry	✓ (identical)
`spans[].start_time`	ISO 8601 UTC	ISO 8601 UTC	✓ (identical)
`spans[].duration_ms`	Integer	Integer	✓ (identical)
`spans[].status`	OK/ERROR/BLOCKED/CIRCUIT_OPEN	OK/ERROR	✓ (superset — new statuses)
`spans[].attributes`	Detection + policy metadata	Governance metadata	✓ (extensible by design)
`summary`	Same schema	Same schema	✓ (identical)

Appendix B: Comparison with Existing Adapters

Aspect	Existing OpenAI Adapter	AIGP-Light OpenAI Adapter
External dependency	Governance service (CHECK/RECORD/TRACE)	None
Authentication	HMAC secret + gov_url	None required
Trace destination	Remote governance service	Local file/stdout
Policy source	Remote (CHECK response)	Local YAML
Failure mode	Governance service down → governance disabled	Never disabled (local)
Trust accrual	Immediate (remote)	Deferred (import when upgrading)
Install weight	aigp-client + aigp-agent-core (~100KB)	aigp-light (~50KB)

RFC: AIGP-Light — Minimal Detect & Respond Client

RFC: AIGP-Light — Minimal Detect & Respond Client

1. Purpose

2. Scope Boundaries

2.1 In Scope

2.2 Out of Scope (deferred to full AIGP)

3. Architecture

4.2 Usage Examples

5. Policy Engine

5.1 PolicyEngine Interface

5.2 Policy Configuration Schema

6. Signal Store

6.1 SignalStore Interface

6.2 Windowing Strategy

7. Response Actions

7.1 Action Implementations

7.2 Action Priority and Composition

8. Trace Emitter

8.1 TraceEmitter Interface

8.2 Trace Output Format

8.3 Forward Compatibility

9. Interceptor

9.1 Call Lifecycle

9.2 Signal Extraction

10. Framework Adapters

10.1 Adapter Pattern

10.2 Adapter Interface Compatibility

11. Error Handling

11.1 Exception Hierarchy

11.2 Failure Modes

12. Testing Strategy

12.1 Unit Tests

12.2 Integration Tests

13. AIGP Governance Integration (Optional)

13.1 Trace Compatibility

14. Dependencies

14.1 Runtime Dependencies

14.2 Optional Dependencies

14.3 Development Dependencies

15. Non-Functional Requirements

16. Security Considerations

17. Open Questions

18. Implementation Plan

Appendix A: Trace Format Cross-Reference

Appendix B: Comparison with Existing Adapters

5.1 `PolicyEngine` Interface

6.1 `SignalStore` Interface

8.1 `TraceEmitter` Interface