Team Conversation

Let a writer ask a reviewer for help and pick up the reply, without you relaying it. Use it when members need to talk mid-run: a question, a clarification, a second look. When the shape is fixed in advance, a review panel or a writer and a reviewer is simpler. Each member has a name, a role and a brief. A saved message that mentions a member requests that member’s next turn, and the conversation stays in the run’s record, so a fresh executor reads the messages and the turns they requested.

Shape

The conversation

The file uses the stored graph executor and local functions for the two members, so it needs no model account:

examples/team-conversation.ts (excerpt)

import {
  compileGraph, createGraphExecutor, loadRunDefinition, persistRunDefinition,
  projectTeamRooms, resolveGraphPlan, teamGraphType,
  type DomainEventEnvelope, type GraphNodeBinding, type JsonValue, type ResultContract, type RunStoragePolicy,
  type TeamDefinition, type TeamGraphResult, type TeamMessage, type TeamTurnResult,
} from '@obversa/runtime';
import { createLocalRunStorage } from '@obversa/runtime/storage/local';

The reviewer waits for a question because its initialTurn is false:

examples/team-conversation.ts (excerpt)

const definition: TeamDefinition = {
  id: 'release-team', definitionVersion: 1,
  data: {
    task: 'Prepare a release note.', globalConcurrency: 1, maxTurnsPerMember: 2,
    communication: { rooms: [{ id: 'review', members: ['writer', 'reviewer'] }], tailMessages: 3 },
  },
  nodes: [
    { id: 'writer', data: { role: 'writer', brief: 'Draft the release note.' } },
    { id: 'reviewer', data: { role: 'reviewer', brief: 'Check the draft.', initialTurn: false } },
  ],
  edges: [],
};

task goes to every member, maxTurnsPerMember caps the turns each may take, and communication names the rooms and how many recent messages a member sees on its next turn. Edges stay empty: messages request turns, and edges don’t connect members. The writer sends its question in its successful turn result. mentions holds member IDs; the runtime doesn’t look for names in the text:

examples/team-conversation.ts (excerpt)

  const nodes = {
    writer: binding('writer', join(temporaryRoot, 'writer'), async ({ input }): Promise<TeamTurnResult> => {
      const turn = input as TurnInput;
      if (turn.result === null) return {
        summary: 'Review requested.',
        posts: [{ roomId: 'review', text: `Please review: ${turn.task}`, mentions: ['reviewer'] }],
      };
      const reply = turn.messages.find((message) => message.sender === 'reviewer');
      assert.ok(reply, 'The writer requires the saved reviewer reply.');
      return { summary: `Finished ${turn.task}: ${reply.text}`, data: { replyId: reply.id, reply: reply.text } };
    }),
    reviewer: binding('reviewer', join(temporaryRoot, 'reviewer'), async ({ input }) => {
      const question = (input as TurnInput).messages.find((message) => message.sender === 'writer');
      assert.ok(question, 'The reviewer requires the saved writer question.');
      return {
        summary: 'Draft checked.',
        posts: [{ roomId: question.roomId, text: `Reviewed ${question.id}: ${question.text}`, mentions: ['writer'] }],
      };
    }),
  };

The reviewer receives the saved question. Its reply mentions writer, which gives the writer another turn with that reply. To ask for an answer, a member ends its turn: a message is saved with the member’s successful result, not while an engine is still speaking, and a failed or unfinished turn sends nothing. A mention requests a later turn and never interrupts one already running. Several pending mentions for the same member share one queued turn and keep their triggering messages. Every post is checked against room membership before any post from that turn is accepted, so a member can’t post to a room it isn’t in or mention someone outside it.

What the run did

The example writes a temporary record, reopens it inside the program with a fresh executor, and removes its files before exit. Run it with npx tsx team-conversation.ts:

Output

{
  "messages": [
    {
      "sender": "writer",
      "text": "Please review: Prepare a release note."
    },
    {
      "sender": "reviewer",
      "text": "Reviewed team/writer/1/0: Please review: Prepare a release note."
    }
  ],
  "order": [
    "writer",
    "reviewer",
    "writer"
  ],
  "answers": [
    {
      "name": "writer",
      "summary": "Finished Prepare a release note.: Reviewed team/writer/1/0: Please review: Prepare a release note."
    },
    {
      "name": "reviewer",
      "summary": "Draft checked."
    }
  ],
  "projectionRevision": 10,
  "replayAddedEvents": false,
  "temporaryDirectoryRemoved": true
}

The saved messages, dispatch order and final answers come from the record and the result. The fresh executor adds no events when it reads the completed conversation (replayAddedEvents is false), and the writer’s final answer contains the reviewer’s reply. projectTeamRooms can rebuild each room as a plain text file, one line per accepted message, that people and tools tail like an ordinary chat. This path compiles teamGraphType for the stored executor. It doesn’t create or merge worktrees or run callable review functions for you, so a team result isn’t proof that a merge or review happened. For a callable job with its own workspace and review behaviour, use team(config) as the graph contract describes.

Keep the run record and the room files outside member workspaces, and use the host’s access controls. Room membership controls which posts are accepted and what each member receives; it isn’t a filesystem sandbox, and processes that share unrestricted filesystem access can read the same files.

Full file

examples/team-conversation.ts

import assert from 'node:assert/strict';
import { createHash } from 'node:crypto';
import { access, mkdir, mkdtemp, realpath, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';

// #region imports
import {
  compileGraph, createGraphExecutor, loadRunDefinition, persistRunDefinition,
  projectTeamRooms, resolveGraphPlan, teamGraphType,
  type DomainEventEnvelope, type GraphNodeBinding, type JsonValue, type ResultContract, type RunStoragePolicy,
  type TeamDefinition, type TeamGraphResult, type TeamMessage, type TeamTurnResult,
} from '@obversa/runtime';
import { createLocalRunStorage } from '@obversa/runtime/storage/local';
// #endregion imports

// #region definition
const definition: TeamDefinition = {
  id: 'release-team', definitionVersion: 1,
  data: {
    task: 'Prepare a release note.', globalConcurrency: 1, maxTurnsPerMember: 2,
    communication: { rooms: [{ id: 'review', members: ['writer', 'reviewer'] }], tailMessages: 3 },
  },
  nodes: [
    { id: 'writer', data: { role: 'writer', brief: 'Draft the release note.' } },
    { id: 'reviewer', data: { role: 'reviewer', brief: 'Check the draft.', initialTurn: false } },
  ],
  edges: [],
};
// #endregion definition

const graph = compileGraph(teamGraphType, definition);
// Keys are sorted so these JSON bytes match the contract's schema digest.
const resultSchema = {
  properties: {
    data: {},
    posts: {
      items: {
        properties: {
          mentions: { items: { type: 'string' }, type: 'array' },
          roomId: { type: 'string' },
          text: { type: 'string' },
        },
        required: ['roomId', 'text', 'mentions'],
        type: 'object',
      },
      type: 'array',
    },
    summary: { type: 'string' },
  },
  required: ['summary'],
  type: 'object',
} as const;
const resultRecord = {
  name: 'team-turn', version: 1,
  schemaDigest: `sha256:${createHash('sha256').update(JSON.stringify(resultSchema)).digest('hex')}` as const,
};

const storagePolicy = {
  schemaVersion: 1, maxEventPayloadBytes: 64_000, maxAppendBatchBytes: 128_000,
  maxArtifactBytes: 1_000_000, maxTotalArtifactBytesPerRun: 4_000_000,
  retention: 'until-run-delete',
  sensitiveContent: { marked: 'reject', exact: 'reject', freeText: 'redact-before-hash' },
} as const satisfies RunStoragePolicy;

function binding(member: string, directory: string, runData: NonNullable<GraphNodeBinding['runData']>): GraphNodeBinding {
  const resultContract: ResultContract = {
    record: resultRecord,
    schema: resultSchema,
    validate(value) {
      const issue = graph.validateNodeResult!(member, value as JsonValue);
      if (issue !== null) throw new Error(`${issue.path}: ${issue.message}`);
      return value as TeamTurnResult;
    },
  };
  return {
    prompt: null, scratchDirectory: directory,
    workspace: { mode: 'none', directory: null, allowedPaths: [] },
    trustedCaller: {}, permissions: [],
    policy: {
      inputBytes: 10_000, outputBytes: 10_000, timeoutMs: 5_000,
      teardownGraceMs: 100, memoryBytes: 10_000_000,
      filesChanged: 0, linesChanged: 0, callTokens: null,
    },
    resultContract, runData, parseResult: null, tokenBudget: null,
    decideAction: async () => ({ kind: 'allow' }),
  };
}

const temporaryRoot = await realpath(await mkdtemp(join(tmpdir(), 'obversa-team-conversation-')));
const runId = 'release-team-run';
const openStorage = () => createLocalRunStorage({
  directory: join(temporaryRoot, 'storage'), namespace: 'team-conversation', policy: storagePolicy,
});
let report;
try {
  const storage = openStorage();
  const packageIdentity = {
    source: 'npm:@example/team-conversation', version: '1.0.0',
    digest: `sha256:${'3'.repeat(64)}` as const,
  };
  await persistRunDefinition(storage, {
    runId, eventId: 'release-team-started', timestamp: '2026-01-01T00:00:00.000Z',
    graphDefinition: graph.definition,
    resolvedPlan: resolveGraphPlan(graph.describe(), {
      package: packageIdentity, admission: { package: packageIdentity, permissions: [] }, executionLanes: [],
    }),
    resolvedInputs: {}, workspaceBinding: null, hostBinding: null,
  });
  for (const member of ['writer', 'reviewer']) await mkdir(join(temporaryRoot, member));
  type TurnInput = { task: string; result: TeamTurnResult | null; messages: TeamMessage[] };
  // #region posts
  const nodes = {
    writer: binding('writer', join(temporaryRoot, 'writer'), async ({ input }): Promise<TeamTurnResult> => {
      const turn = input as TurnInput;
      if (turn.result === null) return {
        summary: 'Review requested.',
        posts: [{ roomId: 'review', text: `Please review: ${turn.task}`, mentions: ['reviewer'] }],
      };
      const reply = turn.messages.find((message) => message.sender === 'reviewer');
      assert.ok(reply, 'The writer requires the saved reviewer reply.');
      return { summary: `Finished ${turn.task}: ${reply.text}`, data: { replyId: reply.id, reply: reply.text } };
    }),
    reviewer: binding('reviewer', join(temporaryRoot, 'reviewer'), async ({ input }) => {
      const question = (input as TurnInput).messages.find((message) => message.sender === 'writer');
      assert.ok(question, 'The reviewer requires the saved writer question.');
      return {
        summary: 'Draft checked.',
        posts: [{ roomId: question.roomId, text: `Reviewed ${question.id}: ${question.text}`, mentions: ['writer'] }],
      };
    }),
  };
  // #endregion posts
  const executor = await createGraphExecutor({ runId, graph, storage, nodes, engines: [] });
  const result = await executor.run(new AbortController().signal);
  assert.equal(result.kind, 'complete');
  if (result.kind !== 'complete') throw new Error('The conversation did not complete.');
  const output = result.output as TeamGraphResult;
  assert.equal(output.agents.find((agent) => agent.name === 'writer')?.result?.summary,
    'Finished Prepare a release note.: Reviewed team/writer/1/0: Please review: Prepare a release note.',
    'The writer must finish after receiving the reply.');

  const saved: DomainEventEnvelope[] = [];
  const stream = { namespace: storage.record.namespace, streamId: runId };
  for await (const event of storage.eventStore.read(stream)) saved.push(event);
  const order = saved.filter((event) => event.type === 'graph:node-dispatched')
    .map((event) => (event.payload as { nodeId: string }).nodeId);
  assert.deepEqual(order, ['writer', 'reviewer', 'writer']);
  const messages = saved.filter((event) => event.type === 'graph:node-completed').flatMap((event) => {
    const completion = event.payload as { nodeId: string; result: TeamTurnResult };
    return (completion.result.posts ?? []).map((post) => ({ sender: completion.nodeId, text: post.text }));
  });

  // #region projection
  const projection = await projectTeamRooms({ storage, runId, directory: join(temporaryRoot, 'operator-rooms') });
  // #endregion projection
  const reopened = openStorage();
  const loaded = await loadRunDefinition(reopened, runId);
  const replay = await createGraphExecutor({
    runId, graph: compileGraph(teamGraphType, loaded.record.payload.definition.graphDefinition.value as TeamDefinition),
    storage: reopened, nodes, engines: [],
  });
  assert.deepEqual(await replay.run(new AbortController().signal), result);
  const after: DomainEventEnvelope[] = [];
  for await (const event of reopened.eventStore.read(stream)) after.push(event);
  assert.deepEqual(after, saved, 'Completed replay must not add events.');
  report = {
    messages, order,
    answers: output.agents.map((agent) => ({ name: agent.name, summary: agent.result?.summary ?? null })),
    projectionRevision: projection.revision, replayAddedEvents: after.length !== saved.length,
  };
} finally {
  await rm(temporaryRoot, { recursive: true, force: true });
}
const temporaryDirectoryRemoved = await access(temporaryRoot).then(() => false, (error: NodeJS.ErrnoException) => {
  if (error.code !== 'ENOENT') throw error;
  return true;
});
assert.equal(temporaryDirectoryRemoved, true);
console.log(JSON.stringify({ ...report, temporaryDirectoryRemoved }, null, 2));

Next steps

Outside graph types: the team form contract, and team(config) for a callable team with its own workspace.
Runtime: teamGraphType, TeamDefinition, the turn result and room fields, and projectTeamRooms.
The record: where the saved conversation lives and how a fresh executor reads it.

Start

Concepts

Patterns

Examples

Evals and Gates

Recording Runs

Driving Runs

Surfaces and Hosts

Writing Workflow Shapes

Memory

Workspace

Compare

FAQ

API Reference

Team Conversation

Shape

The conversation

What the run did

Next steps

​Shape

​The conversation

​What the run did

​Next steps

Shape

The conversation

What the run did

Next steps