diff --git a/docs/15-milestones.md b/docs/15-milestones.md
index 2768fd9..dcf2f4d 100644
--- a/docs/15-milestones.md
+++ b/docs/15-milestones.md
@@ -67,7 +67,8 @@ signed AEP evidence instead.
 - [ ] `aep/zk-proofs/`: Zero-Knowledge (ZK-SNARK) attestation exporter for privacy-preserving AEP evidence verification (`passport prove --privacy-mode zk`)
 - [ ] `agentbom/policy/`: Embed WebAssembly-native Open Policy Agent (OPA/Rego) evaluator for inline runtime guardrail enforcement
 - [x] `wasmagent-js/spiffe/`: SPIFFE/SPIRE cryptographic identity driver binding Wasm sandbox workloads to enterprise mTLS credentials
-- [ ] `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion pipeline for instant posture drift detection and passport revocation
+- [x] `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion pipeline for instant posture drift detection and passport revocation
+  - Owner: `WasmAgent/trace-pipeline` (evidence-pipeline tier; owns evidence admission / training-data pipeline). Reference surface landed in this hub (`trace-pipeline/stream/stream.ts`) and is tracked by `tests/e2e/trace_pipeline_stream_test.go`; the production gRPC service port lives in the sibling repo.
 - [ ] `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline evidence buffering and eventual ledger synchronization
 - [ ] `agent-golden-path/multi-agent/`: Multi-agent procurement federation workload (buyer copilot ↔ supplier copilot) under signed AEP contracts
 - [ ] `wasmagent-ops/resilience/`: Automated circuit breaker and transactional rollback mechanism triggered on policy violation events
diff --git a/docs/project-index.json b/docs/project-index.json
index 9d6700a..834b0bc 100644
--- a/docs/project-index.json
+++ b/docs/project-index.json
@@ -116,7 +116,7 @@
       "status": "shipped",
       "visibility": "public",
       "in_profile": true,
-      "summary": "evomerge on PyPI — eval_trust paired statistics, AgentTrustScore stable JSON schema, training-data admission gate, wasmagent-js v1.x schema compatible.",
+      "summary": "evomerge on PyPI — eval_trust paired statistics, AgentTrustScore stable JSON schema, training-data admission gate, real-time gRPC telemetry/event ingestion stream with instant posture drift detection and passport revocation (trace-pipeline/stream/), wasmagent-js v1.x schema compatible.",
       "url": "https://github.com/WasmAgent/trace-pipeline",
       "focus": "adjacent"
     },
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 655369b..fe81e18 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -83,7 +83,9 @@ The machine-readable `focus` field in
 ### Evidence pipelines ✅
 
 - ✅ `trace-pipeline` (`evomerge` on PyPI) — eval_trust paired statistics,
-  AgentTrustScore stable JSON schema, training-data admission gate; schema
+  AgentTrustScore stable JSON schema, training-data admission gate, and the
+  real-time gRPC telemetry / event ingestion stream (`trace-pipeline/stream/`)
+  for instant posture drift detection and passport revocation; schema
   compatible with `wasmagent-js` v1.x AEP.
 - 🚧 `wasmagent-train-replay` — **Research / Preview**: causal evidence layer
   for distributed GPU training with no stable API or published package yet:
diff --git a/tests/e2e/trace_pipeline_stream_test.go b/tests/e2e/trace_pipeline_stream_test.go
new file mode 100644
index 0000000..2fea42b
--- /dev/null
+++ b/tests/e2e/trace_pipeline_stream_test.go
@@ -0,0 +1,108 @@
+package e2e
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/WasmAgent/.github/pkg/docs"
+)
+
+// TestTracePipelineStream validates the trace-pipeline/stream reference surface
+// for the Milestone 6 bullet:
+//
+//	trace-pipeline/stream/: Real-time gRPC telemetry and event ingestion
+//	pipeline for instant posture drift detection and passport revocation
+//
+// It checks that:
+//   - trace-pipeline/stream/stream.ts ships a TelemetryIngestPipeline exposing
+//     a gRPC-style streaming connect(), per-frame event ingestion (send),
+//     posture drift detection (onDrift), and Trust Passport revocation
+//     (onRevocation / PassportRevokedError).
+//   - trace-pipeline/stream/stream.test.ts exercises real-time ingestion,
+//     posture drift detection, and passport revocation.
+//   - The project index advertises the stream surface on the trace-pipeline
+//     repository (the owning evidence-pipeline repo).
+//   - The Milestone 6 bullet in docs/15-milestones.md is marked complete so the
+//     hub roadmap tracks the shipped surface.
+func TestTracePipelineStream(t *testing.T) {
+	streamDir := filepath.Join("..", "..", "trace-pipeline", "stream")
+
+	// 1. Reference implementation must ship the pipeline surface.
+	driverPath := filepath.Join(streamDir, "stream.ts")
+	driverSource, err := os.ReadFile(driverPath)
+	if err != nil {
+		t.Fatalf("trace-pipeline/stream/stream.ts is missing: %v", err)
+	}
+	for _, fragment := range []string{
+		"export class TelemetryIngestPipeline",
+		"export class PassportRevokedError",
+		"export interface TelemetryEvent",
+		"export interface PostureBaseline",
+		"export interface DriftSignal",
+		"export interface RevocationSignal",
+		"connect(",
+		"send(",
+		"onDrift",
+		"onRevocation",
+	} {
+		if !strings.Contains(string(driverSource), fragment) {
+			t.Errorf("trace-pipeline/stream stream.ts is missing required capability %q", fragment)
+		}
+	}
+
+	// 2. Reference tests must cover real-time ingestion, drift detection, and
+	// passport revocation.
+	testPath := filepath.Join(streamDir, "stream.test.ts")
+	testSource, err := os.ReadFile(testPath)
+	if err != nil {
+		t.Fatalf("trace-pipeline/stream coverage is missing: %v", err)
+	}
+	for _, scenario := range []string{
+		"ingests compliant telemetry events",
+		"detects posture drift",
+		"revokes the Trust Passport",
+		"PassportRevokedError",
+	} {
+		if !strings.Contains(string(testSource), scenario) {
+			t.Errorf("trace-pipeline/stream test is missing scenario %q", scenario)
+		}
+	}
+
+	// 3. The project index must advertise the stream surface on the owning repo.
+	projectIndex, err := docs.LoadProjectIndex()
+	if err != nil {
+		t.Fatalf("Failed to load project index: %v", err)
+	}
+	tracePipeline, found := projectIndex.GetRepoByName("trace-pipeline")
+	if !found {
+		t.Fatal("trace-pipeline repository not found in project index")
+	}
+	summary := strings.ToLower(tracePipeline.Summary)
+	for _, keyword := range []string{"stream", "telemetry", "drift", "revocation"} {
+		if !strings.Contains(summary, keyword) {
+			t.Errorf("trace-pipeline summary does not mention %q: %s", keyword, tracePipeline.Summary)
+		}
+	}
+
+	// 4. The milestone bullet must be marked complete.
+	milestones, err := os.ReadFile("../../docs/15-milestones.md")
+	if err != nil {
+		t.Fatalf("Failed to read docs/15-milestones.md: %v", err)
+	}
+	bulletFound := false
+	for _, line := range strings.Split(string(milestones), "\n") {
+		if strings.Contains(line, "`trace-pipeline/stream/`") {
+			bulletFound = true
+			if !strings.HasPrefix(strings.TrimSpace(line), "- [x]") {
+				t.Errorf("trace-pipeline/stream milestone bullet is not checked: %s", line)
+			}
+		}
+	}
+	if !bulletFound {
+		t.Error("trace-pipeline/stream milestone bullet not found in docs/15-milestones.md")
+	}
+
+	t.Log("Real-time gRPC telemetry and event ingestion pipeline validated for trace-pipeline")
+}
diff --git a/trace-pipeline/README.md b/trace-pipeline/README.md
new file mode 100644
index 0000000..96ebecd
--- /dev/null
+++ b/trace-pipeline/README.md
@@ -0,0 +1,53 @@
+# trace-pipeline reference surfaces
+
+This directory hosts the hub-tracked reference surfaces for the
+[`WasmAgent/trace-pipeline`](https://github.com/WasmAgent/trace-pipeline)
+repository (evidence-pipeline tier — trace ingestion, evidence admission, and
+training-data pipeline). The production implementation lives in the sibling
+repo; this hub copy is the org-level reference contract exercised by
+`tests/e2e/`.
+
+## `stream/` — Real-time gRPC telemetry and event ingestion pipeline
+
+Milestone 6 bullet:
+
+> `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion
+> pipeline for instant posture drift detection and passport revocation
+
+`stream/stream.ts` ships a dependency-free `TelemetryIngestPipeline`:
+
+- **`connect(baseline, passportId)`** opens a gRPC-style streaming telemetry
+  channel for an agent bound to a declared `PostureBaseline`.
+- **`send(event)`** ingests a `TelemetryEvent` frame and returns the agent's
+  updated `PostureSnapshot` in real time.
+- **`onDrift(callback)`** pushes a `DriftSignal` the instant an event violates
+  the agent's posture baseline.
+- **`onRevocation(callback)`** emits a `RevocationSignal` (revoking the agent's
+  Trust Passport) when drift crosses the configured severity or distinct-kind
+  thresholds; subsequent sends raise `PassportRevokedError`.
+
+```ts
+import { TelemetryIngestPipeline } from "./stream/stream";
+
+const pipeline = new TelemetryIngestPipeline({ revokeSeverityThreshold: 5 });
+const stream = pipeline.connect(
+  {
+    agentId: "agent-1",
+    policyRef: "posture:payroll:1.2",
+    allowedKinds: ["tool.call", "network.read"],
+    maxSeverity: 5,
+  },
+  "passport-1",
+);
+stream.onRevocation((signal) => console.log("revoked", signal.passportId));
+const snapshot = stream.send({
+  agentId: "agent-1",
+  eventId: "evt-42",
+  timestamp: new Date().toISOString(),
+  kind: "data.write",
+  observed: { severity: 5 },
+});
+// snapshot.compliant === false; revocation signal fires immediately
+```
+
+Run the reference tests with `bun test` from this directory.
diff --git a/trace-pipeline/stream/stream.test.ts b/trace-pipeline/stream/stream.test.ts
new file mode 100644
index 0000000..02282d8
--- /dev/null
+++ b/trace-pipeline/stream/stream.test.ts
@@ -0,0 +1,110 @@
+// Real-time gRPC telemetry and event ingestion pipeline tests.
+//
+// Exercises real-time ingestion, instant posture drift detection, and Trust
+// Passport revocation for the Milestone 6 reference surface:
+//
+// > `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion
+// > pipeline for instant posture drift detection and passport revocation
+
+import { describe, expect, it } from "bun:test";
+import {
+  InvalidTelemetryEventError,
+  PassportRevokedError,
+  StreamAlreadyOpenError,
+  TelemetryIngestPipeline,
+  type PostureBaseline,
+  type TelemetryEvent,
+} from "./stream";
+
+const baseline: PostureBaseline = {
+  agentId: "agent-1",
+  policyRef: "posture:payroll:1.2",
+  allowedKinds: ["tool.call", "network.read"],
+  maxSeverity: 5,
+};
+
+function event(overrides: Partial<TelemetryEvent> = {}): TelemetryEvent {
+  return {
+    agentId: "agent-1",
+    eventId: "evt-1",
+    timestamp: "2026-08-03T00:00:00.000Z",
+    kind: "tool.call",
+    observed: {},
+    ...overrides,
+  };
+}
+
+describe("TelemetryIngestPipeline", () => {
+  it("ingests compliant telemetry events in real time", () => {
+    const pipeline = new TelemetryIngestPipeline();
+    const stream = pipeline.connect(baseline, "passport-1");
+    const snapshot = stream.send(event({ eventId: "evt-1" }));
+    expect(snapshot.compliant).toBe(true);
+    expect(snapshot.driftCount).toBe(0);
+    expect(snapshot.revoked).toBe(false);
+    expect(pipeline.openStreams()).toContain("agent-1");
+    stream.close();
+  });
+
+  it("detects posture drift immediately on a non-compliant event", () => {
+    const pipeline = new TelemetryIngestPipeline();
+    const stream = pipeline.connect(baseline, "passport-1");
+    const driftSignals: string[] = [];
+    stream.onDrift((signal) => driftSignals.push(signal.kind));
+    const snapshot = stream.send(
+      event({ eventId: "evt-2", kind: "data.write", observed: { severity: 2 } }),
+    );
+    expect(snapshot.compliant).toBe(false);
+    expect(snapshot.driftCount).toBe(1);
+    expect(snapshot.lastDriftKind).toBe("data.write");
+    expect(driftSignals).toEqual(["data.write"]);
+  });
+
+  it("revokes the Trust Passport when drift severity crosses the threshold", () => {
+    const pipeline = new TelemetryIngestPipeline({ revokeSeverityThreshold: 4 });
+    const stream = pipeline.connect(baseline, "passport-1");
+    const revocations: string[] = [];
+    stream.onRevocation((signal) => revocations.push(signal.passportId));
+    stream.send(
+      event({ eventId: "evt-3", kind: "data.write", observed: { severity: 5 } }),
+    );
+    expect(revocations).toEqual(["passport-1"]);
+    expect(stream.isClosed()).toBe(true);
+  });
+
+  it("revokes the Trust Passport after too many distinct drift kinds accumulate", () => {
+    const pipeline = new TelemetryIngestPipeline({ maxDriftKinds: 2 });
+    const stream = pipeline.connect(baseline, "passport-1");
+    const revocations: string[] = [];
+    stream.onRevocation((signal) => revocations.push(signal.revocationReason));
+    stream.send(event({ eventId: "evt-4", kind: "data.write" }));
+    expect(revocations).toEqual([]);
+    stream.send(event({ eventId: "evt-5", kind: "exec.shell" }));
+    expect(revocations).toHaveLength(1);
+    expect(revocations[0]).toContain("passport-1");
+  });
+
+  it("rejects telemetry after the passport has been revoked", () => {
+    const pipeline = new TelemetryIngestPipeline({ revokeSeverityThreshold: 1 });
+    const stream = pipeline.connect(baseline, "passport-1");
+    stream.send(
+      event({ eventId: "evt-6", kind: "exec.shell", observed: { severity: 3 } }),
+    );
+    expect(() =>
+      stream.send(event({ eventId: "evt-7", kind: "tool.call" })),
+    ).toThrow(PassportRevokedError);
+  });
+
+  it("rejects malformed telemetry events and duplicate open streams", () => {
+    const pipeline = new TelemetryIngestPipeline();
+    const stream = pipeline.connect(baseline, "passport-1");
+    expect(() => stream.send(event({ eventId: "" }))).toThrow(
+      InvalidTelemetryEventError,
+    );
+    expect(() => pipeline.connect(baseline, "passport-2")).toThrow(
+      StreamAlreadyOpenError,
+    );
+    stream.close();
+    expect(pipeline.openStreams()).not.toContain("agent-1");
+  });
+});
diff --git a/trace-pipeline/stream/stream.ts b/trace-pipeline/stream/stream.ts
new file mode 100644
index 0000000..6977b79
--- /dev/null
+++ b/trace-pipeline/stream/stream.ts
@@ -0,0 +1,315 @@
+/**
+ * Real-time gRPC telemetry and event ingestion pipeline.
+ *
+ * Reference surface for the Milestone 6 bullet:
+ *
+ * > `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion
+ * > pipeline for instant posture drift detection and passport revocation
+ *
+ * Dependency-free by design (matching `wasmagent-js/runtime.ts`). Agents open a
+ * streaming telemetry channel (`TelemetryIngestPipeline.connect()`), push typed
+ * telemetry/event frames (`TelemetryStream.send()`), and the pipeline compares
+ * every frame against the agent's declared posture baseline. A non-compliant
+ * frame produces a `DriftSignal` in near real time; once drift crosses the
+ * configured severity or distinct-kind thresholds the pipeline emits a
+ * `RevocationSignal` so the trust network can revoke the agent's Trust Passport
+ * immediately.
+ */
+
+export type AgentId = string;
+export type PassportId = string;
+export type PolicyRef = string;
+
+/** A single telemetry event frame streamed by an agent over the gRPC channel. */
+export interface TelemetryEvent {
+  readonly agentId: AgentId;
+  readonly eventId: string;
+  /** ISO-8601 timestamp captured at emission time. */
+  readonly timestamp: string;
+  /** Event kind, e.g. "tool.call", "network.read", "data.write". */
+  readonly kind: string;
+  /** Optional observed payload; may carry a numeric `severity` (1..maxSeverity). */
+  readonly observed: Record<string, unknown>;
+}
+
+/** The declared posture baseline an agent must hold to. */
+export interface PostureBaseline {
+  readonly agentId: AgentId;
+  readonly policyRef: PolicyRef;
+  /** Event kinds the agent's posture allows; everything else is drift. */
+  readonly allowedKinds: readonly string[];
+  /** Maximum severity a single drift event can carry. */
+  readonly maxSeverity: number;
+}
+
+/** A posture drift signal emitted when an event violates the baseline. */
+export interface DriftSignal {
+  readonly agentId: AgentId;
+  readonly eventId: string;
+  readonly kind: string;
+  readonly policyRef: PolicyRef;
+  readonly severity: number;
+  readonly reason: string;
+  readonly timestamp: string;
+}
+
+/** A passport revocation signal emitted when drift crosses the threshold. */
+export interface RevocationSignal extends DriftSignal {
+  readonly passportId: PassportId;
+  readonly revocationReason: string;
+}
+
+/** Snapshot of an agent's posture after a telemetry event is ingested. */
+export interface PostureSnapshot {
+  readonly agentId: AgentId;
+  readonly compliant: boolean;
+  readonly driftCount: number;
+  readonly lastDriftKind: string | undefined;
+  readonly revoked: boolean;
+}
+
+export class PassportRevokedError extends Error {
+  constructor(agentId: AgentId, passportId: PassportId) {
+    super(`passport ${passportId} for agent ${agentId} has been revoked`);
+    this.name = "PassportRevokedError";
+  }
+}
+
+export class InvalidTelemetryEventError extends Error {
+  constructor(reason: string) {
+    super(`invalid telemetry event: ${reason}`);
+    this.name = "InvalidTelemetryEventError";
+  }
+}
+
+export class StreamAlreadyOpenError extends Error {
+  constructor(agentId: AgentId) {
+    super(`agent ${agentId} already has an open telemetry stream`);
+    this.name = "StreamAlreadyOpenError";
+  }
+}
+
+/**
+ * A real-time streaming telemetry channel for a single agent. Mirrors the
+ * streaming shape of a gRPC bidi `TelemetryStream` RPC: `send()` ingests a
+ * frame, `onDrift`/`onRevocation` deliver push signals, and `close()`
+ * terminates the channel.
+ */
+export interface TelemetryStream {
+  readonly agentId: AgentId;
+  readonly passportId: PassportId;
+  /** Ingest a telemetry event frame and return the agent's updated posture. */
+  send(event: TelemetryEvent): PostureSnapshot;
+  /** Subscribe to drift signals; returns an unsubscribe function. */
+  onDrift(callback: (signal: DriftSignal) => void): () => void;
+  /** Subscribe to passport revocation signals; returns an unsubscribe function. */
+  onRevocation(callback: (signal: RevocationSignal) => void): () => void;
+  /** Close the streaming channel. Sends after close throw. */
+  close(): void;
+  /** Whether the channel has been closed (or revoked). */
+  isClosed(): boolean;
+}
+
+export interface TelemetryIngestPipelineOptions {
+  /** Drift severity at or above which the passport is revoked (default 5). */
+  readonly revokeSeverityThreshold?: number;
+  /** Distinct non-compliant kinds allowed before revocation (default 3). */
+  readonly maxDriftKinds?: number;
+}
+
+/** Extract the drift severity carried by an event, clamped to the baseline max. */
+export function eventSeverity(event: TelemetryEvent, maxSeverity: number): number {
+  const raw = event.observed?.severity;
+  const severity = typeof raw === "number" && Number.isFinite(raw) && raw > 0 ? raw : 1;
+  return Math.min(severity, Math.max(1, maxSeverity));
+}
+
+/** Validate a telemetry event frame against the stream it is sent on. */
+export function validateEvent(event: TelemetryEvent, agentId: AgentId): void {
+  if (!event || typeof event !== "object") {
+    throw new InvalidTelemetryEventError("event must be an object");
+  }
+  if (!event.eventId || typeof event.eventId !== "string") {
+    throw new InvalidTelemetryEventError("eventId is required");
+  }
+  if (event.agentId !== agentId) {
+    throw new InvalidTelemetryEventError(
+      `agentId ${event.agentId} does not match stream agent ${agentId}`,
+    );
+  }
+  if (!event.kind || typeof event.kind !== "string") {
+    throw new InvalidTelemetryEventError("kind is required");
+  }
+  if (!event.timestamp || typeof event.timestamp !== "string") {
+    throw new InvalidTelemetryEventError("timestamp is required");
+  }
+}
+
+class TelemetryStreamImpl implements TelemetryStream {
+  readonly agentId: AgentId;
+  readonly passportId: PassportId;
+
+  private readonly baseline: PostureBaseline;
+  private readonly revokeSeverityThreshold: number;
+  private readonly maxDriftKinds: number;
+  private readonly driftCallbacks = new Set<(signal: DriftSignal) => void>();
+  private readonly revocationCallbacks = new Set<(signal: RevocationSignal) => void>();
+  private readonly driftKinds = new Set<string>();
+  private driftCount = 0;
+  private lastDriftKind: string | undefined;
+  private closed = false;
+  private revoked = false;
+
+  constructor(
+    baseline: PostureBaseline,
+    passportId: PassportId,
+    revokeSeverityThreshold: number,
+    maxDriftKinds: number,
+  ) {
+    this.agentId = baseline.agentId;
+    this.passportId = passportId;
+    this.baseline = baseline;
+    this.revokeSeverityThreshold = revokeSeverityThreshold;
+    this.maxDriftKinds = maxDriftKinds;
+  }
+
+  send(event: TelemetryEvent): PostureSnapshot {
+    if (this.closed) {
+      throw new InvalidTelemetryEventError("stream is closed");
+    }
+    if (this.revoked) {
+      throw new PassportRevokedError(this.agentId, this.passportId);
+    }
+    validateEvent(event, this.agentId);
+
+    const snapshot = (compliant: boolean): PostureSnapshot => ({
+      agentId: this.agentId,
+      compliant,
+      driftCount: this.driftCount,
+      lastDriftKind: this.lastDriftKind,
+      revoked: this.revoked,
+    });
+
+    if (this.baseline.allowedKinds.includes(event.kind)) {
+      return snapshot(true);
+    }
+
+    const severity = eventSeverity(event, this.baseline.maxSeverity);
+    const reason = `event kind "${event.kind}" is not allowed by policy ${this.baseline.policyRef}`;
+    const drift: DriftSignal = {
+      agentId: this.agentId,
+      eventId: event.eventId,
+      kind: event.kind,
+      policyRef: this.baseline.policyRef,
+      severity,
+      reason,
+      timestamp: event.timestamp,
+    };
+    this.driftCount += 1;
+    this.lastDriftKind = event.kind;
+    this.driftKinds.add(event.kind);
+    for (const callback of this.driftCallbacks) {
+      callback(drift);
+    }
+
+    if (
+      severity >= this.revokeSeverityThreshold ||
+      this.driftKinds.size >= this.maxDriftKinds
+    ) {
+      this.revoke(drift);
+    }
+    return snapshot(false);
+  }
+
+  private revoke(trigger: DriftSignal): void {
+    if (this.revoked) return;
+    this.revoked = true;
+    const signal: RevocationSignal = {
+      ...trigger,
+      passportId: this.passportId,
+      revocationReason:
+        `passport ${this.passportId} revoked for agent ${this.agentId}: ` +
+        `posture drift "${trigger.kind}" violates policy ${trigger.policyRef}`,
+    };
+    for (const callback of this.revocationCallbacks) {
+      callback(signal);
+    }
+  }
+
+  onDrift(callback: (signal: DriftSignal) => void): () => void {
+    this.driftCallbacks.add(callback);
+    return () => this.driftCallbacks.delete(callback);
+  }
+
+  onRevocation(callback: (signal: RevocationSignal) => void): () => void {
+    this.revocationCallbacks.add(callback);
+    return () => this.revocationCallbacks.delete(callback);
+  }
+
+  close(): void {
+    this.closed = true;
+  }
+
+  isClosed(): boolean {
+    return this.closed || this.revoked;
+  }
+}
+
+/**
+ * The real-time gRPC telemetry and event ingestion pipeline. Manages one
+ * streaming channel per agent and enforces posture baselines with instant
+ * drift detection and Trust Passport revocation.
+ */
+export class TelemetryIngestPipeline {
+  private readonly revokeSeverityThreshold: number;
+  private readonly maxDriftKinds: number;
+  private readonly streams = new Map<AgentId, TelemetryStreamImpl>();
+
+  constructor(options: TelemetryIngestPipelineOptions = {}) {
+    this.revokeSeverityThreshold = options.revokeSeverityThreshold ?? 5;
+    this.maxDriftKinds = options.maxDriftKinds ?? 3;
+    if (this.revokeSeverityThreshold <= 0) {
+      throw new InvalidTelemetryEventError("revokeSeverityThreshold must be positive");
+    }
+    if (this.maxDriftKinds <= 0) {
+      throw new InvalidTelemetryEventError("maxDriftKinds must be positive");
+    }
+  }
+
+  /** Open a real-time gRPC telemetry channel for an agent's posture baseline. */
+  connect(baseline: PostureBaseline, passportId: PassportId): TelemetryStream {
+    if (!baseline || !baseline.agentId) {
+      throw new InvalidTelemetryEventError("baseline.agentId is required");
+    }
+    if (
+      !baseline.policyRef ||
+      !baseline.allowedKinds ||
+      !Array.isArray(baseline.allowedKinds)
+    ) {
+      throw new InvalidTelemetryEventError(
+        "baseline.policyRef and baseline.allowedKinds are required",
+      );
+    }
+    const existing = this.streams.get(baseline.agentId);
+    if (existing && !existing.isClosed()) {
+      throw new StreamAlreadyOpenError(baseline.agentId);
+    }
+    const stream = new TelemetryStreamImpl(
+      baseline,
+      passportId,
+      this.revokeSeverityThreshold,
+      this.maxDriftKinds,
+    );
+    this.streams.set(baseline.agentId, stream);
+    return stream;
+  }
+
+  /** Agent IDs that currently have an open telemetry stream. */
+  openStreams(): AgentId[] {
+    const open: AgentId[] = [];
+    for (const [agentId, stream] of this.streams) {
+      if (!stream.isClosed()) open.push(agentId);
+    }
+    return open;
+  }
+}