# Real-time chat integration API

How to connect a third-party AI chat application to the Observatory so that every
message is analysed as the conversation happens.

This is the practical companion to [`api.md`](api.md) (endpoint reference) and
[`integration-guide.md`](integration-guide.md) (the ownership model). Read this
one if you are building the integration.

**Base path:** `/api/v1` · **Content type:** `application/json` · **Transport for
results:** Server-Sent Events

---

## 1. The mental model

Three things to internalise before writing code, because they shape the whole
integration:

**Your application never owns the data.** Every thread belongs to a human data
owner. Your application acts *on behalf of* that owner through an explicit
delegation in your token. This is why there is no "API key writes to my
account" mode — there is no account for your application to write to.

**Ingestion is synchronous, analysis is not.** `POST /turns` returns as soon as
Layer 1 (deterministic, per-turn measures) is computed. Speech acts and Layer 2
(LCC / ICCC) are queued and arrive later over SSE. Do not block your chat UI on
analysis.

**`null` means "not measurable", never zero.** A `null` latency is not a fast
response; it means you did not send the timestamps needed to compute it. Never
coerce `null` to `0` — it corrupts every average downstream.

```
   your chat app                      Observatory
   ────────────                       ───────────
   user sends message  ──────────▶  POST /turns          → 201 + Layer 1
   assistant streams…
   assistant completes ──────────▶  POST /turns          → 201 + Layer 1
                                          │
                                          ├─ queued: speech acts
                                          └─ queued: LCC / ICCC
   SSE subscriber      ◀──────────  analysis.core.completed
                       ◀──────────  analysis.speech_act.completed
                       ◀──────────  iccc.updated
```

---

## 2. Quick start

```bash
BASE=https://your-observatory.example.org/api/v1
TOKEN=…            # delegated token, see §3
```

**Create the thread once per conversation.** Supply your own `threadId` so the
call is idempotent.

```bash
curl -X POST "$BASE/threads" \
  -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  -d '{
    "threadId": "conv_8f21",
    "title": "Support conversation",
    "interactionType": "general_dyad",
    "consentVersion": "your-consent-v3",
    "sourceApplicationId": "your-app",
    "participants": [
      {"participantId": "user", "type": "human", "roles": ["peer"]},
      {"participantId": "bot",  "type": "assistant", "roles": ["assistant"],
       "model": "claude-sonnet-5", "provider": "anthropic"}
    ]
  }'
```

**Post each message as it completes.**

```bash
curl -X POST "$BASE/threads/conv_8f21/turns" \
  -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  -d '{
    "turnId": "msg_8412",
    "participantId": "user",
    "speakerType": "human",
    "text": "How does cache eviction work?",
    "submittedAt": "2026-04-14T09:02:11.480Z"
  }'
```

**Subscribe to results.**

```
GET /threads/conv_8f21/events        (text/event-stream)
```

That is the entire hot path. Everything below is detail.

---

## 3. Authentication

### Production (OIDC)

Send a bearer token. To act for a data owner, the token needs **both**:

| Requirement | Where |
|---|---|
| `application_client` in the roles claim | `roles` / `role` claim |
| The owner's subject | `act.sub` (RFC 8693) **or** `on_behalf_of` claim |

A delegation claim *without* the `application_client` role is rejected with
`401 delegation_forbidden` — this is deliberate, so a leaked user token cannot
be replayed as an application.

```
Authorization: Bearer <token>
```

The token's `act.sub` identifies the owner; the Observatory derives an internal
owner key from it. You never send the owner id in the body.

### Development (mock mode)

When the deployment runs `AUTH_MODE=mock` (never permitted in production —
configuration refuses to start), identity comes from headers:

```
X-Mock-Subject: demo-owner
X-Mock-Roles: application_client
X-Mock-On-Behalf-Of: end-user-42
```

Useful for local integration work before your OIDC provider is wired up.

---

## 4. Mapping a streaming chat to turn ingestion

This is the part most integrations get wrong.

### Post once per completed message, not per token

A "turn" is a complete contribution. Do **not** POST partial deltas while the
assistant is still streaming — you would create hundreds of turns for one
message and destroy every structural measure. Stream to your user as you
normally would, and post once when the message is finished.

What you capture *during* the stream still matters — it becomes the timing
fields on that single POST:

```ts
// in your chat app
const submittedAt = new Date().toISOString();      // user hit send / request issued
let firstTokenAt: string | undefined;
let lastTokenAt: string | undefined;

for await (const delta of llmStream) {
  firstTokenAt ??= new Date().toISOString();       // first visible token
  render(delta);
}
lastTokenAt = new Date().toISOString();            // stream closed

await postTurn({
  turnId: message.id,
  participantId: 'bot',
  speakerType: 'assistant',
  text: fullText,
  submittedAt, firstTokenAt, lastTokenAt,
  model: 'claude-sonnet-5',
  provider: 'anthropic',
  finishReason: 'end_turn',
  usage: providerUsage,                            // the provider's own numbers
  metadata: { streaming: true, clientRequestId: requestId },
});
```

### Order matters, and is your responsibility

Turns are sequenced in arrival order. Post the user's message before the
assistant's reply to that message. If you fan out ingestion across workers you
can reorder a conversation — either post sequentially per thread, or use the
batch endpoint (§5) which preserves the order of the array.

### Do not block the user on ingestion

Ingestion is fast, but it is still a network call to another service. Enqueue
it; do not `await` it inside your render path. If the Observatory is
unreachable, your chat must keep working — buffer and retry (§8).

### What each field unlocks

| Field | If you omit it |
|---|---|
| `usage.visibleOutputTokens` | Token counts fall back to a local estimate flagged `MISSING_PROVIDER_USAGE`; ER, AR, HS, AHR lose accuracy |
| `usage.totalApiInputTokens` | API Amplification Ratio is `null` |
| `firstTokenAt` / `lastTokenAt` | TTFT, generation duration, E2E and TPOT are all `null` |
| `submittedAt` | The server timestamps on arrival, so latency includes your network delay |
| `explicitTargetParticipantIds` | Reply targets are inferred at lower confidence, weakening Responsivity and Social Impact |
| `parentTurnId` | Reply structure is unavailable to target resolution |
| `model` / `provider` | Per-model comparisons are unavailable |

Send the provider's own usage numbers. Do not compute your own token counts and
label them as provider figures — provenance is recorded and inspected.

---

## 5. Endpoints you will use

| Method | Path | Purpose |
|---|---|---|
| `POST` | `/threads` | Create a conversation. Idempotent when you supply `threadId`. |
| `POST` | `/threads/:threadId/participants` | Add a participant mid-conversation |
| `POST` | `/threads/:threadId/turns` | **Ingest one message** — the hot path |
| `POST` | `/threads/:threadId/turns/batch` | Ingest 1–500 messages in order |
| `GET`  | `/threads/:threadId/events` | **SSE stream** of analysis results |
| `GET`  | `/threads/:threadId/analysis` | Thread-level Layer 1 + speech-act aggregate |
| `GET`  | `/threads/:threadId/analysis/turns/:turnId` | Per-turn measures + speech acts |
| `GET`  | `/threads/:threadId/lcc` | Learner curves (when configured) |
| `GET`  | `/threads/:threadId/iccc` | Collaboration curves (when configured) |
| `GET`  | `/threads/:threadId/turns` | Read back stored contributions |
| `GET`  | `/api/v1/meta` | Algorithm and schema versions (unauthenticated) |

### `POST /threads/:threadId/turns`

Request fields (`*` = required):

| Field | Type | Notes |
|---|---|---|
| `participantId` * | identifier | Must already exist on the thread |
| `speakerType` * | enum | `human` `assistant` `agent` `avatar` `system` `tool` |
| `turnId` | identifier | **Supply this.** Your message id. Makes the call idempotent. |
| `text` | string | ≤ 200 000 chars |
| `parentTurnId` | identifier | The message being replied to |
| `sequence` | int ≥ 0 | Usually omit; assigned server-side |
| `explicitTargetParticipantIds` | identifier[] | ≤ 64. Who this turn addresses |
| `submittedAt` | ISO 8601 | When the turn was issued |
| `firstTokenAt` / `lastTokenAt` | ISO 8601 | Streaming boundaries |
| `model` / `provider` | string | ≤ 120 chars |
| `finishReason` | string | ≤ 64 chars |
| `usage` | object | `newInputTokens` `totalApiInputTokens` `outputTokens` `visibleOutputTokens` `cachedInputTokens` `reasoningTokens` |
| `metadata` | object | `streaming` `toolCalls` `locale` `pageFocused` `clientRequestId`; extra keys allowed |

An **identifier** is `[A-Za-z0-9][A-Za-z0-9_-]{0,127}` — opaque, no PII. A
violation returns `400 unsafe_identifier`.

Response — `201` for a new turn, `200` for an idempotent replay:

```json
{
  "created": true,
  "contribution": { "turnId": "msg_8412", "sequence": 4, "…": "…" },
  "analysis": {
    "layer1": { "…": "per-turn measures, already computed" },
    "layer1Status": "completed",
    "layer2Status": "queued",
    "speechActStatus": "queued"
  },
  "jobIds": ["job_…"]
}
```

`layer1` is available immediately. The two `queued` statuses resolve over SSE.

### `POST /threads/:threadId/turns/batch`

```json
{ "turns": [ { "…": "same shape as above" } ] }
```

1–500 turns, ingested **sequentially** so ordering and sequence assignment are
preserved. Returns `{ results: [...], count: n }`, with `201` if any turn was
newly created. Use this for backfill or to flush a retry buffer — not for the
live path, where one-at-a-time keeps latency measures honest.

---

## 6. Receiving results in real time (SSE)

```
GET /api/v1/threads/:threadId/events
Accept: text/event-stream
Authorization: Bearer <token>
```

Authorization is enforced **before** the stream opens; an unauthorized caller
never receives a stream. The connection is bound to one thread.

### Frame format

Standard SSE. Every frame carries an `id:` (use it for resume), an `event:`
naming the type, and a JSON `data:` payload:

```
id: evt_7e0a4b123b214d56a8fe8f0e72e39d03
event: turn.ingested
data: {"eventId":"evt_7e0a…","type":"turn.ingested","threadId":"thr_2c53…",
       "occurredAt":"2026-07-31T04:40:03.041Z",
       "data":{"turnId":"msg_1","sequence":0,"participantId":"user","speakerType":"human"}}
```

The internal owner key is stripped from the payload — it is never transmitted.

### Event catalogue

| Event | Fires when | Typical reaction |
|---|---|---|
| `connected` | Stream opens. Carries `clientId`, `replayBufferSize` | Note it; nothing to do |
| `turn.ingested` | A contribution was stored | Optimistically mark the message analysed |
| `analysis.core.completed` | Layer 1 finished for a turn | Refresh per-turn measures |
| `analysis.speech_act.started` | Classification queued | Show a pending indicator |
| `analysis.speech_act.completed` | Speech acts available | Refresh acts / structural measures |
| `analysis.speech_act.failed` | Classification failed | Leave the prior value; do not show zero |
| `semantic.started` | Layer 2 began for a turn | Pending indicator |
| `semantic.completed` | Layer 2 finished | Refresh LCC / ICCC |
| `lcc.updated` | Learner curves changed | Refresh LCC view |
| `iccc.updated` | Collaboration curves changed | Refresh ICCC view |
| `iccc.social_impact.revised` | A provisional Social Impact became final | **Re-read; the earlier value was provisional** |
| `analysis.failed` | An analysis job failed | Surface degraded state |
| `thread.metrics.updated` | Thread-level aggregates changed | Refresh header stats |
| `export.completed` | An export finished | Offer the download |
| `grant.updated` | A research grant changed | Refresh sharing UI |
| `heartbeat` | Periodic keepalive | Reset your idle timer |

`iccc.social_impact.revised` deserves attention: Social Impact is genuinely
revisable as later turns arrive. If you cache it, this event is your
invalidation signal.

### Reconnecting without losing events

The hub keeps a replay buffer (size reported on `connected`). On reconnect,
send the last id you processed:

```
Last-Event-ID: evt_7e0a4b123b214d56a8fe8f0e72e39d03
```

or, when you cannot set headers, `?lastEventId=evt_…`. Events after that id are
replayed. Track the last id you *processed*, not the last you received.

### ⚠️ Browser `EventSource` cannot authenticate in production

The `EventSource` API cannot set an `Authorization` header, and this deployment
accepts identity from query parameters **only in mock mode**. So a browser
cannot subscribe directly to a production stream.

**Consume the stream in your backend** and relay to your own frontend over your
existing socket:

```ts
// your server — can set headers
const res = await fetch(`${BASE}/threads/${threadId}/events`, {
  headers: { Authorization: `Bearer ${token}`, Accept: 'text/event-stream' },
});
for await (const event of parseSse(res.body!)) {
  relayToBrowser(threadId, event);   // your own transport
}
```

This is also the right boundary for authorization: your backend already knows
which of your users may see which conversation.

---

## 7. Enabling Layer 2 (LCC / ICCC)

Layer 2 is **never enabled implicitly**. Requesting an engine requires its
configuration block on the thread:

```json
{
  "interactionType": "tutorial_dyad",
  "semanticConfiguration": {
    "enabled": true,
    "mode": "lcc",
    "interactionType": "tutorial_dyad",
    "lcc": {
      "referenceModelId": "rm_newton",
      "learnerParticipantIds": ["user"],
      "tutorParticipantIds": ["bot"],
      "classificationMethod": "hybrid"
    }
  }
}
```

Preconditions are enforced, not assumed. A two-person chat does not qualify for
LCC however long it runs: LCC needs an instructional `interactionType`, a
designated learner, a tutor role, and a versioned reference model. When the
preconditions are unmet the endpoint returns `status: "not_applicable"` **with
reasons** rather than fabricating a curve. Treat that as a valid answer, not an
error.

`mode` is `none` | `lcc` | `iccc` | `lcc_and_iccc`. Requesting a mode without
its configuration block is a `400`.

---

## 8. Reliability: idempotency, retries, ordering

**Idempotency is keyed on thread + `turnId`.** Retrying with the same `turnId`
returns `200` with the stored record and does not duplicate. Always send a
stable `turnId` derived from your own message id — this is what makes at-least-
once delivery safe.

**Recommended client behaviour:**

| Situation | Do |
|---|---|
| `5xx` / network failure | Retry with backoff, same `turnId` |
| `429` | Back off per the rate-limit headers, then retry |
| `409 concurrent_modification` | Retry the request — a manifest write raced |
| `400 validation_failed` | **Do not retry.** Fix the payload; inspect `error.details` |
| `401` / `403` | Do not retry. Refresh the token or fix the delegation |
| Observatory unreachable | Buffer locally, flush later via `/turns/batch` |

Because ingestion is idempotent, a buffer flush that partially succeeded is safe
to replay in full.

**Ordering:** post sequentially per thread. Concurrency *across* threads is fine
and encouraged; concurrency *within* one thread can reorder a conversation.

---

## 9. Errors

Every error uses one envelope, and echoes the request id (also returned as the
`X-Request-Id` header) so you can correlate with server logs:

```json
{
  "error": {
    "code": "validation_failed",
    "message": "Request validation failed.",
    "details": [{ "path": "participantId", "message": "must be an opaque identifier…" }],
    "requestId": "req_357d09bdbe9a45ae9329d7b02be119c2"
  }
}
```

| Status | `code` | Meaning |
|---|---|---|
| 400 | `validation_failed` | Body/query failed schema validation; see `details` |
| 400 | `unsafe_identifier` | An id violated the identifier pattern |
| 401 | `unauthenticated` | Missing/invalid token |
| 401 | `delegation_forbidden` | Delegation claim without `application_client` |
| 403 | `owner_required` / `role_required` | Authenticated but not permitted |
| 403 | `cors_origin_denied` | Origin not in the allowlist |
| 404 | `not_found` | Unknown thread, turn or export |
| 409 | `concurrent_modification` | Optimistic-concurrency clash; retry |
| 409 | `incompatible_semantic_model` | Embedding space changed; re-analysis needed |
| 413 | — | Body exceeded the size limit |
| 429 | `rate_limited` | Too many requests |
| 500 | `internal_error` | Server fault; the message is deliberately generic |

---

## 10. Limits

| Limit | Default | Configured by |
|---|---|---|
| Request body | 1 MB | `REQUEST_BODY_LIMIT` |
| Rate limit | 300 requests / minute | `RATE_LIMIT_MAX_REQUESTS`, `RATE_LIMIT_WINDOW_MINUTES` |
| Turn text | 200 000 chars | schema |
| Turns per batch | 500 | schema |
| Participants per thread | 64 | schema |
| Identifier length | 128 chars | schema |

Confirm the deployment's actual values with its operator — all are configurable.

---

## 11. Reference client

A minimal, production-shaped integration: non-blocking ingestion with a retry
buffer, plus a resuming SSE subscriber.

```ts
type Turn = {
  turnId: string; participantId: string; speakerType: 'human' | 'assistant';
  text?: string; submittedAt?: string; firstTokenAt?: string; lastTokenAt?: string;
  model?: string; provider?: string; usage?: Record<string, number>;
  metadata?: Record<string, unknown>;
};

export class ObservatoryClient {
  private readonly pending: Turn[] = [];
  private flushing = false;

  constructor(
    private readonly base: string,
    private readonly token: () => Promise<string>,
    private readonly threadId: string,
  ) {}

  private async headers() {
    return { 'Content-Type': 'application/json', Authorization: `Bearer ${await this.token()}` };
  }

  /** Never throws and never blocks the chat path. */
  ingest(turn: Turn): void {
    this.pending.push(turn);
    void this.flush();
  }

  private async flush(): Promise<void> {
    if (this.flushing || this.pending.length === 0) return;
    this.flushing = true;
    try {
      while (this.pending.length > 0) {
        const turn = this.pending[0]!;
        const res = await fetch(`${this.base}/threads/${this.threadId}/turns`, {
          method: 'POST', headers: await this.headers(), body: JSON.stringify(turn),
        });

        if (res.ok) { this.pending.shift(); continue; }          // 200 replay or 201 new
        if (res.status === 400 || res.status === 401 || res.status === 403) {
          // Not retryable — drop it rather than blocking every later turn.
          console.error('[observatory] dropping turn', turn.turnId, await res.text());
          this.pending.shift();
          continue;
        }
        // 429 / 409 / 5xx — retry this same turn later, preserving order.
        const retryAfter = Number(res.headers.get('retry-after') ?? 2);
        setTimeout(() => void this.flush(), retryAfter * 1000);
        return;
      }
    } finally {
      this.flushing = false;
    }
  }

  /** Resuming SSE subscription. Run this in your backend, not the browser. */
  async subscribe(onEvent: (type: string, data: unknown) => void, signal?: AbortSignal) {
    let lastId: string | undefined;
    while (!signal?.aborted) {
      try {
        const res = await fetch(`${this.base}/threads/${this.threadId}/events`, {
          headers: {
            ...(await this.headers()),
            Accept: 'text/event-stream',
            ...(lastId ? { 'Last-Event-ID': lastId } : {}),
          },
          signal,
        });
        if (!res.ok || !res.body) throw new Error(`stream failed: ${res.status}`);

        const reader = res.body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';
        for (;;) {
          const { done, value } = await reader.read();
          if (done) break;
          buffer += decoder.decode(value, { stream: true });

          // Frames are separated by a blank line.
          let split: number;
          while ((split = buffer.indexOf('\n\n')) !== -1) {
            const frame = buffer.slice(0, split);
            buffer = buffer.slice(split + 2);

            let id: string | undefined, type = 'message', data = '';
            for (const line of frame.split('\n')) {
              if (line.startsWith('id: ')) id = line.slice(4);
              else if (line.startsWith('event: ')) type = line.slice(7);
              else if (line.startsWith('data: ')) data += line.slice(6);
            }
            if (type === 'heartbeat') continue;
            onEvent(type, data ? JSON.parse(data) : null);
            // Advance only after the handler ran, so a crash replays the event.
            if (id) lastId = id;
          }
        }
      } catch (error) {
        if (signal?.aborted) return;
        console.warn('[observatory] stream dropped; reconnecting', error);
      }
      await new Promise((r) => setTimeout(r, 2000));   // reconnect backoff
    }
  }
}
```

Wiring it into a streaming chat:

```ts
const obs = new ObservatoryClient(BASE, getToken, threadId);

obs.subscribe((type, event) => {
  if (type === 'analysis.core.completed') refreshTurn(event);
  if (type === 'iccc.social_impact.revised') invalidateIcccCache(event);
});

// user turn
obs.ingest({ turnId: userMsg.id, participantId: 'user', speakerType: 'human',
             text: userMsg.text, submittedAt: userMsg.sentAt });

// assistant turn, after the stream completes
obs.ingest({ turnId: reply.id, participantId: 'bot', speakerType: 'assistant',
             text: reply.text, submittedAt, firstTokenAt, lastTokenAt,
             model, provider, usage: reply.usage,
             metadata: { streaming: true, clientRequestId: reply.requestId } });
```

---

## 12. Integration checklist

- [ ] Token carries `application_client` **and** `act.sub` / `on_behalf_of`
- [ ] `threadId` and `turnId` are stable ids from your own system
- [ ] Identifiers match `[A-Za-z0-9][A-Za-z0-9_-]{0,127}` and contain no PII
- [ ] One POST per completed message — never per streamed token
- [ ] `submittedAt`, `firstTokenAt`, `lastTokenAt` captured during streaming
- [ ] Provider's own `usage` forwarded, not re-derived
- [ ] Ingestion is off the render path and cannot break the chat
- [ ] Retries reuse the same `turnId`; `4xx` is not retried
- [ ] Turns posted sequentially within a thread
- [ ] SSE consumed server-side with `Last-Event-ID` resume
- [ ] `iccc.social_impact.revised` invalidates cached Social Impact
- [ ] `null` rendered as "not measurable", never `0`
- [ ] `not_applicable` from LCC/ICCC handled as a valid answer

---

## See also

- [`api.md`](api.md) — full endpoint reference
- [`openapi.yaml`](openapi.yaml) — machine-readable spec
- [`integration-guide.md`](integration-guide.md) — ownership and consent model
- [`metric-dictionary.md`](metric-dictionary.md) — what every measure means, and its prohibited interpretations
- [`authorization-matrix.md`](authorization-matrix.md) — who can read what
- [`lcc-guide.md`](lcc-guide.md) · [`iccc-guide.md`](iccc-guide.md) — Layer 2 engines
- [`privacy-security.md`](privacy-security.md) — data handling obligations