Inside a DyTopo round

Open one round of Dynamic Topology Routing: agents declare what they Need and Offer, cosine similarity wires them into a comm graph, jurisdiction-ineligible agents are dropped, then everyone runs an agentic loop whose findings are gated, debated, verified, and rolled up into memory.
src/dytopo/engine.ts443 lines · DyTopoEngine L72–443
Outline 10 symbolsAgentBillingCtx interface export
DyTopoOptions interface export
DyTopoEngine class exportconstructor method
runRound method
recruitAgents method
fetchAgentMemories method
buildCommGraph method
routeMessages method
persistRoundMemory method
1// SPDX-License-Identifier: AGPL-3.0-only
2// Copyright (C) 2026 Discover Legal
3// This program is free software: you can redistribute it and/or modify it
4// under the terms of the GNU Affero General Public License as published by
5// the Free Software Foundation, either version 3 of the License, or
6// (at your option) any later version. See <https://www.gnu.org/licenses/>.
7
8/**
* DyTopo Engine — Dynamic Topology Routing for Multi-Agent Reasoning
*
* Based on arXiv:2602.06039 (Lu et al., 2026).
* Each reasoning round:
*   1. Manager issues a RoundGoal (natural language).
*   2. Recruited agents emit Need + Offer descriptors conditioned on the goal.
*   3. Need/Offer embeddings are matched via cosine similarity.
*   4. A sparse directed communication graph is constructed (edges above threshold).
*   5. Messages are routed along edges (offering agent's content → needing agent's context).
*   6. Agents process their context and produce findings.
*   7. Round state is written to inter-round memory.
*
* Extended beyond the paper: agents are recruited from a live vector DB (AgentRegistry)
* based on semantic match against the round goal, not from a fixed roster.
*/
24
25import { v4 as uuidv4 } from "uuid";
26import { Config } from "../config.js";
27import { embed, embedBatch, cosineSimilarity } from "../embeddings.js";
28import { logger } from "../logger.js";
29import { auditLogger, ACTOR_SYSTEM } from "../audit/index.js";
30import { Agent } from "../agents/base.js";
31import { AgentRegistry } from "../agents/registry.js";
32import { globalToolRegistry } from "../tools/index.js";
33import { IntraRoundMemoryStore, InterRoundMemoryStore } from "../memory/index.js";
34import { getProvider, resolveModelId } from "../providers/index.js";
35import { selectModel } from "../routing/model.js";
36import { agentLearning } from "../learning/index.js";
37import type { KnowledgeStore } from "../knowledge/index.js";
38import type { TimeStore } from "../time/index.js";
39import type {
AgentDefinition,
AgentMessage,
CommunicationEdge,
Finding,
NeedDescriptor,
OfferDescriptor,
RoundGoal,
RoundState,
Task,
ToneProfile,
50} from "../types.js";
51import { jurisdictionMatch } from "./jurisdiction.js";
52
53/** Billing context forwarded from the orchestrator into each agent's process() call. */
54export interface AgentBillingCtx {
timeStore: TimeStore;
responsibleLawyerId?: string;
responsibleLawyerName?: string;
matterNumber?: string;
clientNumber?: string;
60}
61
62export { jurisdictionMatch } from "./jurisdiction.js";
63
64export interface DyTopoOptions {
registry: AgentRegistry;
memory: InterRoundMemoryStore;
knowledge: KnowledgeStore;
/** Agents pre-selected for this round (e.g. tier-0 root is always included) */
pinnedAgents?: AgentDefinition[];
70}
71
72export class DyTopoEngine {
private readonly registry: AgentRegistry;
private readonly memory: InterRoundMemoryStore;
private readonly knowledge: KnowledgeStore;
private readonly pinnedAgents: AgentDefinition[];
77
constructor(opts: DyTopoOptions) {
  this.registry = opts.registry;
  this.memory = opts.memory;
  this.knowledge = opts.knowledge;
  this.pinnedAgents = opts.pinnedAgents ?? [];
}
84
/**
 * Execute one round of DyTopo orchestration.
 * Returns the completed RoundState including all messages, edges, and findings.
 */
async runRound(task: Task, goal: RoundGoal, lawyerTone?: ToneProfile, billingCtx?: AgentBillingCtx): Promise<RoundState> {
  const roundId = uuidv4();
  const intraMemory = new IntraRoundMemoryStore(roundId);
92
  auditLogger.write({
    event: "round.start",
    actorId: ACTOR_SYSTEM,
    taskId: task.id,
    data: { round: goal.round, phase: goal.phase, roundId },
  });
99
  logger.info("DyTopo round starting", {
    taskId: task.id,
    round: goal.round,
    phase: goal.phase,
    goal: goal.description.slice(0, 80),
  });
106
  // ── Step 1: Recruit agents ──────────────────────────────────────────────
  const recruitedAgents = await this.recruitAgents(goal, task);
  const agentMap = new Map<string, AgentDefinition>();
  for (const a of [...this.pinnedAgents, ...recruitedAgents]) agentMap.set(a.id, a);
111
  const activeDefinitions = Array.from(agentMap.values())
    .filter((a) => jurisdictionMatch(a, task.jurisdiction))
    .slice(0, Config.dytopo.maxAgentsPerRound);
  const activeAgents = activeDefinitions.map((d) => new Agent(d));
116
  logger.info("Agents recruited for round", {
    round: goal.round,
    agents: activeDefinitions.map((a) => a.name),
  });
121
  // ── Step 2: Retrieve inter-round memory for each agent ─────────────────
  const agentMemories = await this.fetchAgentMemories(activeDefinitions, task, goal);
124
  // ── Step 3: Need/Offer descriptors ─────────────────────────────────────
  const needsOffers = await Promise.all(
    activeAgents.map((agent) =>
      agent.generateNeedOffer({
        roundGoal: goal,
        incomingMessages: [],
        memoryEntries: agentMemories.get(agent.definition.id) ?? [],
        taskDescription: task.description,
      }),
    ),
  );
  const needs  = needsOffers.map((no) => no.need);
  const offers = needsOffers.map((no) => no.offer);
138
  // ── Step 4: Build sparse directed comm graph ────────────────────────────
  const edges = await this.buildCommGraph(needs, offers, activeDefinitions);
  logger.info("Communication graph built", {
    round: goal.round,
    edges: edges.length,
    threshold: Config.dytopo.similarityThreshold,
  });
146
  // ── Step 5: Route messages along edges ─────────────────────────────────
  const messages = this.routeMessages(edges, offers, goal.round);
  for (const msg of messages) intraMemory.recordMessage(msg.to, msg);
150
  // ── Step 6: Agents process — full agentic loops ─────────────────────────
  // allSettled, not all: one agent throwing (e.g. a transient provider error)
  // must not discard every other agent's findings and fail the whole round.
  // Per-agent wall-clock cap so one hung agent can't stall the whole round.
  const withTimeout = <T>(p: Promise<T>, agentId: string): Promise<T> =>
    Promise.race([
      p,
      new Promise<T>((_, rej) =>
        setTimeout(
          () => rej(new Error(`Agent ${agentId} exceeded round timeout`)),
          Config.agents.roundTimeoutMs,
        ).unref?.(),
      ),
    ]);
165
  const settled = await Promise.allSettled(
    activeAgents.map((agent) =>
      withTimeout(agent.process({
        roundGoal: goal,
        incomingMessages: intraMemory.getMessagesFor(agent.definition.id),
        memoryEntries: agentMemories.get(agent.definition.id) ?? [],
        taskDescription: task.description,
        taskId: task.id,
        toolRegistry: globalToolRegistry,
        knowledge: this.knowledge,
        memory: this.memory,
        ownerId: task.createdByProfileId,
        assignedLawyerTone: lawyerTone,
        // Agent billing context — only present when a billing rate is configured
        timeStore: billingCtx?.timeStore,
        responsibleLawyerId: billingCtx?.responsibleLawyerId,
        responsibleLawyerName: billingCtx?.responsibleLawyerName,
        matterNumber: billingCtx?.matterNumber,
        clientNumber: billingCtx?.clientNumber,
      }), agent.definition.id),
    ),
  );
  const allFindings = settled.flatMap((r, i) => {
    if (r.status === "fulfilled") return r.value;
    logger.warn("Agent failed during round — skipping its findings", {
      agentId: activeAgents[i].definition.id,
      round: goal.round,
      error: (r.reason as Error)?.message,
    });
    return [];
  });
197
  for (const finding of allFindings) {
    finding.round = goal.round;
    intraMemory.recordFinding(finding.agentId, finding);
    // Write to the intra-round whiteboard so persistRoundMemory can roll it up
    intraMemory.addSharedContext(
      `[${finding.agentName}] ${finding.content.replace(/\s+/g, " ").slice(0, 200)}`,
    );
  }
206
  // ── Step 7: Haiku rollup → inter-round memory ───────────────────────────
  await this.persistRoundMemory(task, goal, allFindings, intraMemory);
209
  const state: RoundState = {
    roundId,
    goal,
    activeAgentIds: activeDefinitions.map((a) => a.id),
    edges,
    messages,
    findings: allFindings,
    status: "complete",
    startedAt: new Date(),
    completedAt: new Date(),
  };
221
  auditLogger.write({
    event: "round.complete",
    actorId: ACTOR_SYSTEM,
    taskId: task.id,
    data: {
      round: goal.round,
      phase: goal.phase,
      roundId,
      agents: activeDefinitions.map((a) => ({ id: a.id, name: a.name })),
      findings: allFindings.length,
      edges: edges.length,
    },
  });
235
  logger.info("DyTopo round complete", {
    round: goal.round,
    findings: allFindings.length,
    messages: messages.length,
  });
241
  return state;
}
244
// ─── Private helpers ────────────────────────────────────────────────────────
246
private async recruitAgents(goal: RoundGoal, task: Task): Promise<AgentDefinition[]> {
  const phaseQueries: Record<string, { tier?: 1 | 2 | 3 }> = {
    intake: { tier: 1 },
    research: { tier: 2 },
    analysis: { tier: 2 },
    drafting: { tier: 2 },
    review: { tier: 2 },
    verification: { tier: 2 },
    delivery: { tier: 1 },
  };
  const tierOpt = phaseQueries[goal.phase] ?? {};
  const topK = Config.dytopo.maxAgentsPerRound - 1;
259
  // From prior rounds in this task, collect agents whose findings were not
  // challenged (positive) and agents whose findings were challenged (negative).
  // Use these to bias recruitment toward historically effective agents.
  const positive: string[] = [];
  const negative: string[] = [];
  for (const round of task.rounds) {
    for (const f of round.findings) {
      if (f.challenged) negative.push(f.agentId);
      else positive.push(f.agentId);
    }
  }
271
  // Deduplicate and cap to avoid blowing out the recommend() request.
  const uniquePositive = [...new Set(positive)].slice(0, 8);
  const uniqueNegative = [...new Set(negative)].slice(0, 4);
275
  // Semantic search — always run to ensure broad, relevant candidate pool.
  const candidates = uniquePositive.length
    ? await this.registry.recommend(goal.description, {
        positive: uniquePositive,
        negative: uniqueNegative,
        ...tierOpt,
        topK,
      })
    : await this.registry.search(goal.description, { ...tierOpt, topK });
285
  // Q-learning rerank — promotes agents with strong historical performance
  // for this exact (phase, jurisdiction, workflowType) combination.
  // The learning layer uses epsilon-greedy exploration so it keeps discovering
  // new agents even as it exploits known good ones.
  const rankedIds = agentLearning.rankCandidates(
    goal.phase,
    task.jurisdiction,
    task.workflowType,
    candidates.map((a) => a.id),
  );
296
  // Restore AgentDefinition objects in the new ranked order.
  const byId = new Map(candidates.map((a) => [a.id, a]));
  return rankedIds.map((id) => byId.get(id)).filter((a): a is AgentDefinition => a != null);
}
301
private async fetchAgentMemories(
  agents: AgentDefinition[],
  task: Task,
  goal: RoundGoal,
): Promise<Map<string, import("../types.js").MemoryEntry[]>> {
  const map = new Map<string, import("../types.js").MemoryEntry[]>();
  await Promise.all(
    agents.map(async (agent) => {
      const entries = await this.memory.query(goal.description, {
        taskId: task.id,
        agentId: agent.id,
        beforeRound: goal.round,
        topK: 6,
      });
      // Also fetch task-level summaries
      const taskEntries = await this.memory.query(goal.description, {
        taskId: task.id,
        beforeRound: goal.round,
        topK: 4,
      });
      map.set(agent.id, [...entries, ...taskEntries]);
    }),
  );
  return map;
}
327
private async buildCommGraph(
  needs: NeedDescriptor[],
  offers: OfferDescriptor[],
  agents: AgentDefinition[],
): Promise<CommunicationEdge[]> {
  // Embed all descriptors in batch
  const needTexts = needs.map((n) => n.text);
  const offerTexts = offers.map((o) => o.text);
  const allTexts = [...needTexts, ...offerTexts];
337
  const embeddings = await embedBatch(allTexts);
339
  const needEmbeddings = embeddings.slice(0, needs.length).map((e) => e.embedding);
  const offerEmbeddings = embeddings.slice(needs.length).map((e) => e.embedding);
342
  const edges: CommunicationEdge[] = [];
  const threshold = Config.dytopo.similarityThreshold;
345
  for (let i = 0; i < needs.length; i++) {
    for (let j = 0; j < offers.length; j++) {
      // An agent does not route messages to itself
      if (needs[i].agentId === offers[j].agentId) continue;
      const sim = cosineSimilarity(needEmbeddings[i], offerEmbeddings[j]);
      if (sim >= threshold) {
        edges.push({
          from: offers[j].agentId,   // offering agent → sends to needing agent
          to: needs[i].agentId,
          similarity: sim,
          offerText: offers[j].text,
        });
      }
    }
  }
361
  // Sort edges by similarity descending for cleaner logs
  return edges.sort((a, b) => b.similarity - a.similarity);
}
365
private routeMessages(
  edges: CommunicationEdge[],
  offers: OfferDescriptor[],
  round: number,
): AgentMessage[] {
  const offerMap = new Map(offers.map((o) => [o.agentId, o.text]));
  return edges.map((edge) => ({
    id: uuidv4(),
    from: edge.from,
    to: edge.to,
    content: `[Offer from ${edge.from}] ${(offerMap.get(edge.from) ?? "").slice(0, 500)}`,
    round,
    timestamp: new Date(),
  }));
}
381
/**
 * Persist intra-round findings as individual memory entries, then synthesize
 * a round-level rollup via Haiku. The rollup is a 2-3 sentence digest of the
 * round's key conclusions — much richer than a string truncation.
 */
private async persistRoundMemory(
  task: Task,
  goal: RoundGoal,
  findings: Finding[],
  intraMemory: IntraRoundMemoryStore,
): Promise<void> {
  // Write individual finding memories in parallel
  await Promise.all(
    findings.map((f) =>
      this.memory.writeFindingMemory({
        taskId: task.id,
        round: goal.round,
        phase: goal.phase,
        agentId: f.agentId,
        finding: f,
      }),
    ),
  );
405
  // Build the round rollup — Haiku synthesis of all findings, falling back to
  // the naive concatenation if the model call fails so memory always gets written.
  let summaryContent: string;
  if (findings.length) {
    const bulletList = findings
      .slice(0, 12)
      .map((f) => `- [${f.agentName}] ${f.content.replace(/\s+/g, " ").slice(0, 150)}`)
      .join("\n");
    try {
      const model = selectModel({ tier: 3, type: "tool", taskType: "descriptor" });
      const provider = getProvider(model);
      const response = await provider.chat({
        model: resolveModelId(model),
        maxTokens: 300,
        system: "You are a legal analysis synthesizer. Produce a concise inter-round memory digest.",
        messages: [{
          role: "user",
          content: `Round ${goal.round} (${goal.phase}) findings:\n${bulletList}\n\nSummarise the key legal conclusions from this round in 2-3 sentences. Be specific — name parties, statutes, or doctrines where present. This summary will be retrieved as memory by agents in the next round.`,
        }],
      });
      const textBlock = response.content.find((b) => b.type === "text");
      summaryContent = textBlock?.type === "text" ? textBlock.text.trim() : bulletList;
    } catch {
      summaryContent = `Round ${goal.round} key findings: ${findings.slice(0, 3).map((f) => f.content.slice(0, 100)).join("; ")}`;
    }
  } else {
    summaryContent = `Round ${goal.round} (${goal.phase}): No findings produced.`;
  }
434
  await this.memory.writeRoundSummary({
    taskId: task.id,
    round: goal.round,
    phase: goal.phase,
    summary: summaryContent,
    findingCount: findings.length,
  });
}
443}
No results