Observe Calls

   the agent writes                the call log                  anybody reads
   ────────────────                ────────────                  ─────────────
   call.log(name, value)  ──┐                                ┌──►  new EventSource(url)      browser, zero libraries
   what the session         ├──►  seq 1, 2, 3, …  ──────────►├──►  useCall / useAgentCalls   @pinecall/web/log/react
   observed (transcripts,   │     append-once,               ├──►  pc.observe({ agent })     Node / your backend
   tools, turns, summary) ──┘     replayable, sealed         └──►  curl -N                   anything that speaks HTTP
                                  by call.summary

• The agent writes. Everything the session observes is appended for you; call.log(name, value) is how your facts get in (Custom entries).
• The log stamps. One fact = one entry = one monotonic seq, per call. seq is the cursor, the dedupe key and the resume key, all at once.
• Anybody reads. Live tail, late join, reconnect, replay and history are the same request with a different after=. Observation is a GET; no WebSocket is opened unless you ask for one to send something.

The wire itself — envelope, vocabulary, headers, framing, filters — is The Call Log. This page is how you consume it.

1. Mint a stream token#

Observation uses a stream token, minted server-side with your API key, so the key never reaches the browser. The token carries the agent set its holder may see, sealed into the signature — a browser cannot widen it.

// your backend — POST /api/token
const { token, server } = await pc.createToken("stream", "dental-desk", undefined, {
  scope: "observe",                       // read-only: no supervise verbs
});
// → { token: "str_…", server: "https://voice.pinecall.io", expiresIn: 3600 }

agentId may be a list — pc.createToken("stream", ["acme-support", "acme-sales"]) — which is how one dashboard watches several agents and how a tenant is scoped to exactly its own: see Multi-tenant dashboards.

A token covers a call iff the call's agent is in its set. Holding another tenant's call id grants nothing — the read is a 403.

2. From the browser#

Zero libraries: new EventSource(url)#

The GET cursor answers Accept: text/event-stream with a held-open SSE body, and that is the header EventSource sends by itself. Each entry is one event: id: is its seq, event: is its type, data: is the full envelope.

<script>
  const es = new EventSource(
    `${server}/v1/calls/${callId}/events?token=${token}&after=0`,
  );

  es.addEventListener("user.message", (e) => {
    const { data } = JSON.parse(e.data);          // the envelope
    if (data.final) addLine("caller", data.text);
  });
  es.addEventListener("bot.speaking", (e) => addLine("bot", JSON.parse(e.data).data.text));
  es.addEventListener("tool.call",   (e) => addChip(JSON.parse(e.data).data.name));

  es.addEventListener("custom", (e) => {          // what call.log() wrote
    const entry = JSON.parse(e.data);
    if (entry.ephemeral) return;                  // live-only, never state
    const { name, value } = entry.data;
    if (name === "appointment.booked") showBooking(value);
  });

  es.addEventListener("log.caught_up", () => setStatus("live"));   // backlog done
  es.addEventListener("call.summary", () => es.close());           // always the last entry
</script>

That is the whole client. Reconnects are free: the browser resends the last id: it saw as Last-Event-ID, the server reads it as after=, and you lose nothing and see nothing twice. The first frame of every body is retry: 1000, so a dropped stream comes back in about a second instead of the browser default.

Why the token is in the query string. EventSource cannot set an Authorization header. The endpoint accepts ?token= for exactly that; the library clients send both. A stream token is short-lived and read-only — that is the trade it is designed for.

With the library: observe()#

@pinecall/web/log is framework-free: the reducer (CallLogView) plus the pipes that feed it. observe() is the one to use — it is sse, degrading to poll if the stream never opens.

import { createCallLogView, observe } from "@pinecall/web/log";

const view = createCallLogView();
view.subscribe(() => render(view.state));         // messages, toolCalls, turns, custom, phase

const obs = observe(view, {
  call: callId, token, server,
  onEntry: (entry, state) => {},                  // every applied entry, in seq order
  onFinish: ({ reason, lastSeq }) => {},          // "summary" | "closed" | "error"
});
obs.kind      // "sse" — what is actually carrying entries
obs.close();

observe(view, opts) never opens a WebSocket on its own. transport is "auto" by default and "auto" means sse → poll; "ws" happens only when you ask for it, and only to send verbs.

With React: useCall / useAgentCalls#

import { useAgentCalls, useCall } from "@pinecall/web/log/react";

// 1 — which calls exist, which are live (the AGENT log: lifecycle only)
const { calls, live } = useAgentCalls("dental-desk", { token, server });

// 2 — one call's content (the CALL log)
const s = useCall<{ "appointment.booked": Booking }>({
  call: live[0]?.call, token, server,
  onCustom: (name, value, _entry) => {            // three params — see the warning
    if (name === "appointment.booked") toast(value.reference);
  },
});

s.messages    // [{ seq, role, text, interim?, speaking?, … }]
s.toolCalls   // [{ id, name, args, result?, ms?, done }]
s.custom      // [{ name, id, value, seq, ts, turn? }] — upserted by (name, id)
s.turns · s.phase · s.live · s.caughtUp · s.transport

Both hooks return view (the reducer itself, for view.entries() or a second pipe) and close(); useCall also returns send(verb).

The options, with the defaults the code actually has (@pinecall/web/log/react, src/log/react.tsx):

⚠️ onCustom narrows only at exact arity. useCall<Custom> types the callback as one parameter tuple per name, which is what lets if (name === "appointment.booked") narrow value. TypeScript only contextually types a union of parameter tuples against a matching arity, so declare all three parameters — (name, value, _entry) => …. Drop _entry and value silently widens back to unknown.

3. From Node — your backend, a worker, a CLI (pc.observe())#

pc.observe() is the same reader, server-side. It opens the same GET with Accept: text/event-stream, feeds the same CallLogView reducer, and gives you three ways to consume it — for await, on(), and the reduced state.

const obs = pc.observe({ agent: "dental-desk", types: ["custom", "call.ended"] });

obs.on("custom", (name, value, entry) => crm.record(entry.call, name, value));
obs.on("entry", (entry, state) => {});            // fires synchronously, never dropped
obs.on("finish", ({ reason, lastSeq }) => {});

for await (const entry of obs) {                  // the only lossy surface — see below
  if (entry.type === "call.started") console.log("call", entry.call);
}

await obs.done;                                   // resolves once; never rejects
obs.close();                                      // idempotent; never reconnects after

Minting needs an agent. The token defaults to one minted with the client's API key (createToken({ channel: "stream", scope: "observe" })), and a stream token's visibility is an agent set — so observe({ call }) without a token must also pass agent. The SDK will not resolve a call id to its agent behind your back, because the only endpoint that would answer needs the very token being minted. Pass { call, agent }, or pass a token you already hold.

Backpressure is a bounded queue. The async iterator buffers at most queueLimit entries — default 1024 — and on overflow the oldest are dropped, never the newest (a slow tail should show recent truth). Every drop is counted in obs.dropped.

const obs = pc.observe({ agent, queueLimit: 4096 });
// …later
if (obs.dropped > 0) console.warn(`iterator skipped ${obs.dropped} entries`);

state is not affected: every entry is reduced into the view before it is queued, so the reduced state is complete even when the iterator skipped rows. on("entry") is likewise never dropped — it fires synchronously. The queue is the only lossy surface, and only under genuine backpressure.

Catching up after your consumer was down#

Persist lastSeq and hand it back as after. That is the whole recovery protocol — the property a webhook cannot give you:

const from = Number(await redis.get(`cursor:${agent}`) ?? 0);
const obs = pc.observe({ agent, after: from });

obs.on("entry", async (entry) => {
  await handle(entry);
  await redis.set(`cursor:${agent}`, String(entry.seq));   // after the work, not before
});

Nothing between from and now is lost: the backlog replays first, log.caught_up marks the boundary, and the tail continues from there. An agent log never seals, so this consumer runs forever.

4. From a terminal — or any language#

No SDK, no browser. One header:

curl -N -H 'Accept: text/event-stream' \
        -H "Authorization: Bearer $TOKEN" \
        "https://voice.pinecall.io/v1/calls/$CALL/events?after=0"

retry: 1000

id: 1
event: call.started
data: {"seq":1,"ts":1786537580.11,"call":"CA…","agent":"dental-desk","type":"call.started","ephemeral":false,"data":{"direction":"inbound","from":"+34…","to":"+1…","channel":"phone"}}

: ping

id: 12
event: user.message
data: {"seq":12,"ts":1786537584.38,"call":"CA…","agent":"dental-desk","type":"user.message","ephemeral":false,"data":{"id":"u1","text":"I need an appointment","final":true}}

Resume with -H "Last-Event-ID: 12", or with ?after=12 — they mean the same thing and the larger of the two wins. Drop the Accept header and the same URL answers a JSON page instead; that is the polling cursor.

5. Resume and failure#

Everything here is one idea: the cursor is the protocol.

6. Filters: types= and durable=1#

Both GET cursors take server-side filters. They are applied at the reader's sink — never on the append path — and seq is never renumbered, so the cursor still means what it means.

# only what your app wrote, plus the terminators
curl -N -H 'Accept: text/event-stream' \
  "$server/v1/calls/$CALL/events?token=$T&types=custom,tool.result&durable=1"

pc.observe({ agent, types: ["custom"], durable: true });
useCall({ call, token, server, types: ["user.message", "bot.speaking"] });

• types=a,b,c — comma-separated, at most 32 names, each matching ^[a-z0-9_.-]{1,64}$. Only the shape is refused: an unknown name is accepted and matches nothing, so a newer client may name a type an older server has never heard of.
• durable=1 — drop ephemeral entries (partial transcripts, word timings, call.log(…, { ephemeral: true })) from the live tail.

The always-pass set ignores both filters, always:

log.gap · log.caught_up · call.ended · call.summary

Those are the markers that say where you stand and when it is over. Without them a types=tool.call client would tail a dead call forever.

Sending verbs is the one WebSocket#

Observation is a read and reads do not need a socket. The only reason to open one is to send — the supervise verbs (whisper, barge in, take over):

const s = useCall({ call, token, server, transport: "ws" });   // opt in, explicitly
s.send({ verb: "whisper", text: "the 10:30 slot is gone" });   // false if not open

That takes WS /v1/attach?token=…&call=<id>&after=<seq>, and a supervise token rather than an observe one:

const t = await pc.createToken("stream", SLUG, undefined, { scope: "supervise", callId });

The verbs (SuperviseVerb, from @pinecall/web/log):

Each one lands in the log as its own entry (supervisor.said, supervisor.whispered, handoff.*) — the audit trail is the same log everybody else is reading. See Live listening and Human takeover.

If you are only watching, leave transport alone. auto never opens a socket, and that is a guarantee, not a default: a dashboard that only reads costs the voice server one HTTP body.

Coming from agent.stream() / agent.ws()?#

They are gone — removed, not deprecated, along with createEventStream(), EventStream and the whole @pinecall/sdk stream/ module. They were in-process: they could only ever see calls handled by that process, they had no cursor, no replay and no history, and they died with the process. The call log has all four, from anywhere.

- app.get("/events", (req, res) => agent.stream(res));
- const es = new EventSource("/events");
+ app.post("/api/token", async (req, res) =>
+   res.json(await pc.createToken("stream", SLUG, undefined, { scope: "observe" })));
+ const { calls, live } = useAgentCalls(SLUG, { token, server });

pc.stream() — the client-level, multi-agent, in-process SSE bus that the run console is built on — still exists. It is a development and single-process convenience, not the observation model: no cursor, no replay, no history, one process only. If you are building something a user will see, read the log.

What's next#

• The Call Log — the wire: envelope, vocabulary, endpoints, framing
• Build a live call app — the whole thing, working, in one app
• Multi-tenant dashboards — one token per tenant, sealed agent set
• Live listening · Human takeover — when you also need to send