Big Michael in a channel

Trace an @BigMichael mention from Teams or Slack: the shared dispatcher parses the command, answers instantly or dispatches async work to the orchestrator, and proactive notifications post results back when the matter task completes.
src/orchestrator.ts1137 lines · Orchestrator L133–1137
Outline 30 symbolsPHASE_SEQUENCES const
recordOrchestratorCost function
parseJsonObject function
Orchestrator class exportconstructor method
init method
submitTask method
getTask method
listTasks method
deleteTask method
assignLawyers method
listTemplates method
submitFromTemplate method
approveGate method
rejectGate method
loadExternalAgents method
loadMikeOSSWorkflows method
loadLavernWorkflows method
loadPlugins method
emit method
runTask method
recordAgentOutcomes method
runPhase method
generateRoundGoal method
synthesise method
tabulate method
buildSourceTextMap method
persistTasks method
restoreTasks method
waitForGates 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/**
* Top-level orchestrator — ties the full system together.
*
* Lifecycle per task:
*   init → plan phases → for each phase: run DyTopo rounds → apply protocols → gate check → next phase
*   → final synthesis
*
* The Root Orchestrator agent (tier 0) generates all RoundGoals via Claude.
* The DyTopo engine assembles the agent graph per round from the registry.
* Findings flow through the debate + verification protocols before final output.
*/
19
20import { EventEmitter } from "events";
21import { readdir, readFile, writeFile, rename } from "fs/promises";
22import { join, extname } from "path";
23import { v4 as uuidv4 } from "uuid";
24import { Config } from "./config.js";
25import { logger } from "./logger.js";
26import { getProvider, resolveModelId } from "./providers/index.js";
27import { selectModel, shouldUseThinking } from "./routing/model.js";
28import { auditLogger, ACTOR_SYSTEM } from "./audit/index.js";
29import { AgentRegistry } from "./agents/registry.js";
30import { Agent } from "./agents/base.js";
31import { ROOT_ORCHESTRATOR, ALL_AGENT_DEFINITIONS } from "./agents/definitions.js";
32import { SettingsStore } from "./settings/index.js";
33import { ProfileStore } from "./auth/index.js";
34import { ClientStore } from "./clients/index.js";
35import { TimeStore } from "./time/index.js";
36import { OcgStore } from "./ocg/index.js";
37import { PreBillStore } from "./billing/prebill.js";
38import { agentLearning } from "./learning/index.js";
39import { DyTopoEngine } from "./dytopo/engine.js";
40import type { AgentBillingCtx } from "./dytopo/engine.js";
41import { DocketMonitor } from "./dockets/monitor.js";
42import { InterRoundMemoryStore } from "./memory/index.js";
43import { KnowledgeStore } from "./knowledge/index.js";
44import { TemplateStore } from "./templates/store.js";
45import { LavernAdapter, LavernWorkflowAdapter, instantiateTemplate, fromExternalConfig, fromMikeOSSWorkflow, sanitizePromptContent } from "./adapters/lavern.js";
46import type { TaskTemplate, ExternalAgentConfig, MikeOSSWorkflow } from "./adapters/lavern.js";
47import { pluginRegistry } from "./adapters/plugin.js";
48import {
applyCitationGate,
runDebate,
runVerificationPipeline,
identifyGateRequests,
53} from "./protocols/index.js";
54import { detectNosLegal } from "./services/classifier.js";
55import { costStore, calcCostUsd, calcWattHours } from "./cost/index.js";
56import { jobQueue } from "./queue/index.js";
57import { isOllamaModel, isLocalModel } from "./providers/index.js";
58import { BudgetMonitor } from "./budget/index.js";
59import { BudgetPredictor } from "./budget/predictor.js";
60import { ConflictGraph } from "./graph/conflict.js";
61import { RegPulseMonitor } from "./regulatory/pulse.js";
62import { DeadlineEngine } from "./deadlines/engine.js";
63import { CitationEngine } from "./citations/engine.js";
64import { MatterHealthMonitor } from "./matters/health.js";
65import { PlaybookStore, PlaybookBuilder } from "./playbook/index.js";
66import { InvoiceValidator } from "./billing/invoice-validator.js";
67import { RedlineEngine } from "./redline/engine.js";
68import { HeadnoteEngine } from "./headnotes/engine.js";
69import { BriefingEngine } from "./briefing/index.js";
70import { PrecedentGenerator } from "./precedent/generator.js";
71import type {
Task,
WorkflowType,
TaskPhase,
RoundGoal,
TaskTable,
77} from "./types.js";
78
79const PHASE_SEQUENCES: Record<WorkflowType, TaskPhase[]> = {
counsel:        ["intake", "research", "drafting", "delivery"],
roundtable:     ["intake", "research", "analysis", "drafting", "review", "delivery"],
adversarial:    ["intake", "research", "analysis", "review", "verification", "delivery"],
review:         ["intake", "analysis", "review", "verification", "delivery"],
tabulate:       ["intake", "analysis", "delivery"],
full_bench:     ["intake", "research", "analysis", "drafting", "review", "verification", "delivery"],
// Lavern workflow types
legal_design:   ["intake", "research", "analysis", "drafting", "review", "delivery"],
pre_engagement: ["intake", "research", "analysis", "delivery"],
89};
90
91/**
* Best-effort extraction of a single JSON object from an LLM response.
* Strips markdown fences and isolates the outermost {...} before parsing.
*/
95function recordOrchestratorCost(
response: import("./providers/index.js").ChatResponse,
modelId: string,
context: import("./cost/index.js").CostContext,
taskId?: string,
100): void {
const isLocal = isOllamaModel(modelId) || isLocalModel(modelId);
const bare = resolveModelId(modelId);
const cw = response.usage.cacheWriteTokens ?? 0;
const cr = response.usage.cacheReadTokens ?? 0;
costStore.record({
  model: bare,
  provider: isLocal ? (isOllamaModel(modelId) ? "ollama" : "local") : "anthropic",
  inputTokens: response.usage.inputTokens,
  outputTokens: response.usage.outputTokens,
  ...(cw ? { cacheWriteTokens: cw } : {}),
  ...(cr ? { cacheReadTokens: cr } : {}),
  costUsd: isLocal ? null : calcCostUsd(bare, response.usage.inputTokens, response.usage.outputTokens, cw, cr),
  estimatedWh: isLocal ? calcWattHours(Config.local.inferenceWatts, response.durationMs) : null,
  estimatedWatts: isLocal ? Config.local.inferenceWatts : null,
  durationMs: response.durationMs,
  context,
  taskId,
});
119}
120
121function parseJsonObject(text: string): unknown | undefined {
const stripped = text.replace(/```(?:json)?/gi, "").trim();
const start = stripped.indexOf("{");
const end = stripped.lastIndexOf("}");
if (start === -1 || end === -1 || end <= start) return undefined;
try {
  return JSON.parse(stripped.slice(start, end + 1));
} catch {
  return undefined;
}
131}
132
133export class Orchestrator {
readonly registry: AgentRegistry;
readonly memory: InterRoundMemoryStore;
readonly knowledge: KnowledgeStore;
readonly templates: TemplateStore;
readonly settings: SettingsStore;
readonly profiles: ProfileStore;
readonly clients: ClientStore;
readonly time: TimeStore;
readonly ocg: OcgStore;
readonly budgetMonitor: BudgetMonitor;
readonly budgetPredictor: BudgetPredictor;
readonly preBills: PreBillStore;
readonly conflictGraph: ConflictGraph;
readonly regPulse = new RegPulseMonitor();
readonly docketMonitor: DocketMonitor;
readonly deadlines = new DeadlineEngine();
readonly citations = new CitationEngine();
readonly matterHealth = new MatterHealthMonitor();
readonly playbookStore: PlaybookStore;
readonly playbookBuilder = new PlaybookBuilder();
readonly invoiceValidator = new InvoiceValidator();
readonly redline = new RedlineEngine();
readonly headnotes = new HeadnoteEngine();
readonly briefing = new BriefingEngine();
readonly precedents = new PrecedentGenerator();
159
private readonly tasks: Map<string, Task> = new Map();
private readonly gateEmitter = new EventEmitter();
readonly progressEmitter = new EventEmitter();
private engine!: DyTopoEngine;
private rootAgent!: Agent;
165
constructor() {
  this.registry = new AgentRegistry();
  this.memory = new InterRoundMemoryStore();
  this.knowledge = new KnowledgeStore();
  this.templates = new TemplateStore();
  this.settings = new SettingsStore();
  this.profiles = new ProfileStore();
  this.clients = new ClientStore();
  this.time = new TimeStore();
  this.ocg = new OcgStore();
  this.budgetMonitor = new BudgetMonitor(this.time, this.clients);
  this.budgetPredictor = new BudgetPredictor();
  this.preBills = new PreBillStore(Config.persistence.preBillsFile);
  this.conflictGraph = new ConflictGraph();
  this.docketMonitor = new DocketMonitor(Config.dockets.file);
  this.playbookStore = new PlaybookStore(Config.persistence.playbooksFile ?? "./data/playbooks.json");
}
183
async init(): Promise<void> {
  // Load persisted admin settings first so they apply before any task runs.
  await this.settings.init();
  await this.profiles.init();
  await this.clients.init();
  await this.time.init();
  await this.ocg.init();
  await this.preBills.init();
  await this.conflictGraph.connect();
  await this.playbookStore.init();
  await agentLearning.init();
  await Promise.all([
    this.registry.init(),
    this.memory.init(),
    this.knowledge.init(),
    this.templates.load(),
  ]);
201
  // Seed agent registry if empty
  const existing = await this.registry.listAll();
  if (!existing.length) {
    logger.info("Seeding agent registry with default agents…");
    await this.registry.registerAll(ALL_AGENT_DEFINITIONS);
  }
208
  // Load external and Lavern agents from filesystem
  await this.loadExternalAgents();
211
  // Load workflow presets — MikeOSS (native format) and Lavern (9 workflow types)
  await this.loadMikeOSSWorkflows();
  await this.loadLavernWorkflows();
215
  // Load generic JSON plugins from adapters/external/ and register their
  // agents, templates, and tools into the live stores.
  await this.loadPlugins();
219
  // Restore persisted tasks and recent audit history into the in-memory buffer
  await this.restoreTasks();
  await auditLogger.restoreFromFile();
223
  this.rootAgent = new Agent(ROOT_ORCHESTRATOR);
  this.engine = new DyTopoEngine({
    registry: this.registry,
    memory: this.memory,
    knowledge: this.knowledge,
    pinnedAgents: [ROOT_ORCHESTRATOR],
  });
231
  // Start regulatory pulse monitor if enabled
  if (this.regPulse.isEnabled()) {
    this.regPulse.start(() => this.listTasks());
    logger.info("RegPulseMonitor started");
  }
237
  // Start docket monitoring if enabled
  await this.docketMonitor.init();
  this.docketMonitor.setKnowledgeStore(this.knowledge);
  if (this.docketMonitor.isEnabled()) {
    this.docketMonitor.start();
    logger.info("DocketMonitor started");
  }
245
  // Load deadline rule files
  await this.deadlines.loadRulesDir(Config.deadlines.rulesDir);
248
  logger.info("Orchestrator ready");
}
251
// ─── Task management ──────────────────────────────────────────────────────
253
private static readonly MAX_CONCURRENT_TASKS = 10;
private static readonly MAX_DESCRIPTION_CHARS = 20_000;
256
async submitTask(params: {
  description: string;
  workflowType: WorkflowType;
  documentIds?: string[];
  clientNumber?: string;
  matterNumber?: string;
  jurisdiction?: string;
  createdByProfileId?: string;
}): Promise<Task> {
  if (params.description.length > Orchestrator.MAX_DESCRIPTION_CHARS) {
    throw new Error(
      `Task description exceeds the ${Orchestrator.MAX_DESCRIPTION_CHARS.toLocaleString()} character limit ` +
      `(${params.description.length.toLocaleString()} received). Please shorten the description.`,
    );
  }
  const running = Array.from(this.tasks.values())
    .filter((t) => t.status === "running" || t.status === "pending").length;
  if (running >= Orchestrator.MAX_CONCURRENT_TASKS) {
    throw new Error(`Server at capacity: ${running} tasks already running. Please wait for one to complete.`);
  }
  const phases = PHASE_SEQUENCES[params.workflowType];
  if (!phases) {
    throw new Error(`Unknown workflowType '${params.workflowType}'. Valid values: ${Object.keys(PHASE_SEQUENCES).join(", ")}`);
  }
  const task: Task = {
    id: uuidv4(),
    description: params.description,
    jurisdiction: params.jurisdiction?.trim().toUpperCase().slice(0, 20) || undefined,
    clientNumber: params.clientNumber?.trim().slice(0, 100) || undefined,
    matterNumber: params.matterNumber?.trim().slice(0, 100) || undefined,
    documentIds: params.documentIds ?? [],
    createdByProfileId: params.createdByProfileId,
    workflowType: params.workflowType,
    status: "pending",
    currentPhase: phases[0],
    currentRound: 0,
    maxRounds: Config.dytopo.maxRounds,
    activeAgentIds: [],
    rounds: [],
    findings: [],
    pendingGates: [],
    createdAt: new Date(),
    updatedAt: new Date(),
  };
301
  // Auto-detect NOSLEGAL taxonomy tags from the description (best-effort, never blocks).
  try {
    task.noslegal = await detectNosLegal(params.description, params.description);
  } catch {
    // detectNosLegal never throws, but guard defensively.
  }
308
  // Open a time entry when a profile is associated with this task.
  if (params.createdByProfileId) {
    const profile = this.profiles.get(params.createdByProfileId);
    if (profile) {
      const entry = this.time.open({
        profileId: profile.id,
        profileName: profile.name,
        taskId: task.id,
        matterNumber: task.matterNumber,
        clientNumber: task.clientNumber,
        description: `Task: ${task.description.slice(0, 200)}`,
        event: "task_run",
        startedAt: new Date(),
      });
      task.activeTimeEntryId = entry.id;
    }
  }
326
  this.tasks.set(task.id, task);
  logger.info("Task submitted", { taskId: task.id, workflow: params.workflowType });
  auditLogger.write({ event: "task.created", actorId: task.createdByProfileId ?? ACTOR_SYSTEM, taskId: task.id, data: { description: params.description, workflowType: params.workflowType } });
  await this.persistTasks();
331
  // Run asynchronously — callers poll getTask() for status
  this.runTask(task).catch((err) => {
    logger.error("Task execution failed", { taskId: task.id, error: err.message });
    task.status = "failed";
    task.error = err.message;
    // Close the time entry even on failure.
    if (task.activeTimeEntryId) {
      const closedOnFail = this.time.close(task.activeTimeEntryId);
      task.activeTimeEntryId = undefined;
      if (closedOnFail) {
        jobQueue.enqueue("summarize_time_entry", {
          entryId: closedOnFail.id,
          taskId: task.id,
          clientNumber: task.clientNumber,
        }).catch((e) => logger.warn("Failed to enqueue summarize job", { error: (e as Error).message }));
      }
    }
    this.emit(task.id, "failed", { error: err.message });
    auditLogger.write({ event: "task.failed", actorId: ACTOR_SYSTEM, taskId: task.id, data: { error: err.message } });
  });
352
  return task;
}
355
getTask(taskId: string): Task | null {
  return this.tasks.get(taskId) ?? null;
}
359
listTasks(): Task[] {
  return Array.from(this.tasks.values());
}
363
/** Delete a matter and its Qdrant memory entries. Returns false if it didn't exist. */
deleteTask(taskId: string): boolean {
  const task = this.tasks.get(taskId);
  if (task) {
    if (task.status === "running" || task.status === "pending" || task.status === "awaiting_gate") {
      throw new Error(`Cannot delete a task in status '${task.status}'; wait for completion or cancel first`);
    }
  }
  const existed = this.tasks.delete(taskId);
  if (existed) {
    this.persistTasks().catch((err) => logger.warn("Failed to persist tasks", { error: err.message }));
    // Clean up orphaned inter-round memory vectors so deleted task data
    // cannot be surfaced by future semantic memory queries.
    this.memory.deleteByTaskId(taskId).catch((err) =>
      logger.warn("Failed to delete task memory from Qdrant", { taskId, error: (err as Error).message }),
    );
    auditLogger.write({ event: "task.deleted", actorId: ACTOR_SYSTEM, taskId, data: {} });
    logger.info("Task deleted", { taskId });
  }
  return existed;
}
385
/** Set the lawyer(s) assigned to a matter (a partner action). */
assignLawyers(taskId: string, lawyerIds: string[], actorId: string = ACTOR_SYSTEM): Task | null {
  const task = this.tasks.get(taskId);
  if (!task) return null;
  const prev = task.assignedLawyerIds ?? [];
  const valid = [...new Set(lawyerIds)].slice(0, 50).filter((id) => this.profiles.get(id));
  task.assignedLawyerIds = valid;
  task.updatedAt = new Date();
  this.persistTasks().catch((err) => logger.warn("Failed to persist tasks", { error: err.message }));
  const added = valid.filter((id) => !prev.includes(id));
  const removed = prev.filter((id) => !valid.includes(id));
  auditLogger.write({ event: "task.assigned", actorId, taskId, data: { lawyerIds: valid, added, removed } });
  return task;
}
400
listTemplates(): TaskTemplate[] {
  return this.templates.list();
}
404
async submitFromTemplate(
  templateId: string,
  substitutions: Record<string, string> = {},
  documentIds?: string[],
  refs?: { clientNumber?: string; matterNumber?: string; createdByProfileId?: string },
): Promise<Task> {
  const template = this.templates.get(templateId);
  if (!template) throw new Error(`Template not found: ${templateId}`);
  const { description, workflowType } = instantiateTemplate(template, substitutions);
  return this.submitTask({ description, workflowType, documentIds, ...refs });
}
416
/**
 * Human approves or rejects a gate request.
 * Approved findings proceed to output; rejected are discarded.
 */
approveGate(taskId: string, gateId: string, note?: string, reviewerProfileId?: string): void {
  const task = this.tasks.get(taskId);
  if (!task) throw new Error(`Task not found: ${taskId}`);
  const gate = task.pendingGates.find((g) => g.id === gateId);
  if (!gate) throw new Error(`Gate not found: ${gateId}`);
  gate.status = "approved";
  gate.reviewerNote = note;
  gate.reviewedAt = new Date();
  task.updatedAt = new Date();
  auditLogger.write({ event: "gate.approved", actorId: reviewerProfileId ?? ACTOR_SYSTEM, taskId, data: { gateId, note } });
431
  // Record a gate_review time entry for the reviewing lawyer.
  if (reviewerProfileId) {
    const profile = this.profiles.get(reviewerProfileId);
    if (profile) {
      const entry = this.time.open({
        profileId: profile.id,
        profileName: profile.name,
        taskId,
        matterNumber: task.matterNumber,
        clientNumber: task.clientNumber,
        description: `Gate review: ${gate.finding.content.slice(0, 100)}`,
        event: "gate_review",
        startedAt: new Date(),
      });
      const closedApprove = this.time.close(entry.id);
      if (closedApprove) {
        jobQueue.enqueue("summarize_time_entry", {
          entryId: closedApprove.id,
          taskId,
          clientNumber: task.clientNumber,
        }).catch((e) => logger.warn("Failed to enqueue summarize job", { error: (e as Error).message }));
      }
    }
  }
456
  this.gateEmitter.emit(`gates:${taskId}`);
  this.persistTasks().catch((err) => logger.warn("Failed to persist tasks", { error: err.message }));
}
460
rejectGate(taskId: string, gateId: string, reason: string, reviewerProfileId?: string): void {
  const task = this.tasks.get(taskId);
  if (!task) throw new Error(`Task not found: ${taskId}`);
  const gate = task.pendingGates.find((g) => g.id === gateId);
  if (!gate) throw new Error(`Gate not found: ${gateId}`);
  gate.status = "rejected";
  gate.reviewerNote = reason;
  gate.reviewedAt = new Date();
  task.findings = task.findings.filter((f) => f.id !== gate.findingId);
  task.updatedAt = new Date();
  auditLogger.write({ event: "gate.rejected", actorId: reviewerProfileId ?? ACTOR_SYSTEM, taskId, data: { gateId, reason } });
472
  // Record a gate_review time entry for the reviewing lawyer.
  if (reviewerProfileId) {
    const profile = this.profiles.get(reviewerProfileId);
    if (profile) {
      const entry = this.time.open({
        profileId: profile.id,
        profileName: profile.name,
        taskId,
        matterNumber: task.matterNumber,
        clientNumber: task.clientNumber,
        description: `Gate review: ${gate.finding.content.slice(0, 100)}`,
        event: "gate_review",
        startedAt: new Date(),
      });
      const closedReject = this.time.close(entry.id);
      if (closedReject) {
        jobQueue.enqueue("summarize_time_entry", {
          entryId: closedReject.id,
          taskId,
          clientNumber: task.clientNumber,
        }).catch((e) => logger.warn("Failed to enqueue summarize job", { error: (e as Error).message }));
      }
    }
  }
497
  this.gateEmitter.emit(`gates:${taskId}`);
  this.persistTasks().catch((err) => logger.warn("Failed to persist tasks", { error: err.message }));
}
501
// ─── External agent loader ────────────────────────────────────────────────
503
private async loadExternalAgents(): Promise<void> {
  const dirs: Array<{ path: string; type: "external" | "lavern" }> = [
    { path: join(process.cwd(), "agents", "external"), type: "external" },
    { path: join(process.cwd(), "agents", "lavern"), type: "lavern" },
  ];
509
  const lavernAdapter = new LavernAdapter();
511
  for (const { path: dir, type } of dirs) {
    let entries: string[];
    try {
      entries = await readdir(dir);
    } catch {
      continue; // directory doesn't exist or isn't readable — skip silently
    }
519
    const defs = [];
    for (const entry of entries) {
      if (extname(entry) !== ".json") continue;
      try {
        const raw = await readFile(join(dir, entry), "utf8");
        const parsed = JSON.parse(raw);
        const items = Array.isArray(parsed) ? parsed : [parsed];
        if (type === "lavern") {
          defs.push(...lavernAdapter.fromConfigs(items));
        } else {
          defs.push(...(items as ExternalAgentConfig[]).map(fromExternalConfig));
        }
      } catch (err) {
        logger.warn("Failed to load external agent file", { file: entry, error: (err as Error).message });
      }
    }
536
    if (defs.length) {
      await this.registry.registerAll(defs);
      logger.info("External agents registered", { source: type, count: defs.length });
    }
  }
}
543
// ─── MikeOSS workflow loader ──────────────────────────────────────────────
545
/**
 * Load MikeOSS workflow presets (native MikeOSSWorkflow format) from
 * workflows/mikeoss/ and register each as a TaskTemplate via the adapter.
 * MikeOSS workflows are task specifications, not agents — our agent system
 * executes them. Files may contain a single workflow or an array.
 */
private async loadMikeOSSWorkflows(): Promise<void> {
  const dir = join(process.cwd(), "workflows", "mikeoss");
  let entries: string[];
  try {
    entries = await readdir(dir);
  } catch {
    return; // directory doesn't exist — skip silently
  }
560
  let loaded = 0;
  for (const entry of entries) {
    if (extname(entry) !== ".json") continue;
    try {
      const raw = await readFile(join(dir, entry), "utf8");
      const parsed = JSON.parse(raw) as MikeOSSWorkflow | MikeOSSWorkflow[];
      const items = Array.isArray(parsed) ? parsed : [parsed];
      for (const wf of items) {
        this.templates.add(fromMikeOSSWorkflow(wf));
        loaded++;
      }
    } catch (err) {
      logger.warn("Failed to load MikeOSS workflow file", { file: entry, error: (err as Error).message });
    }
  }
576
  if (loaded) logger.info("MikeOSS workflows registered as templates", { count: loaded });
}
579
/**
 * Load Lavern workflow definitions from workflows/laverne/ and register each
 * as a TaskTemplate. Lavern workflows specify step pipelines and gate conditions;
 * we map them to our WorkflowType phase sequences.
 */
private async loadLavernWorkflows(): Promise<void> {
  const dir = join(process.cwd(), "workflows", "laverne");
  const adapter = new LavernWorkflowAdapter();
  const templates = await adapter.load(dir);
  for (const t of templates) this.templates.add(t);
  if (templates.length) logger.info("Lavern workflows registered as templates", { count: templates.length });
}
592
// ─── Generic plugin loader ────────────────────────────────────────────────
594
/**
 * Load JSON plugin packages from adapters/external/ via the PluginRegistry.
 * Each package may contribute tools (registered into globalToolRegistry),
 * agent definitions (registered into the AgentRegistry), and workflow
 * templates (added to the TemplateStore).
 */
private async loadPlugins(): Promise<void> {
  const dir = join(process.cwd(), "adapters", "external");
  await pluginRegistry.loadDirectory(dir);
  if (!pluginRegistry.size) return;
605
  const { globalToolRegistry } = await import("./tools/index.js");
607
  const tools = pluginRegistry.allTools();
  for (const tool of tools) globalToolRegistry.register(tool as Parameters<typeof globalToolRegistry.register>[0]);
610
  const agents = pluginRegistry.allAgents();
  if (agents.length) {
    await this.registry.registerAll(agents);
    logger.info("Plugin agents registered", { count: agents.length });
  }
616
  const workflows = pluginRegistry.allWorkflows();
  for (const wf of workflows) this.templates.add(wf);
  if (workflows.length) logger.info("Plugin templates registered", { count: workflows.length });
}
621
// ─── Internal task runner ─────────────────────────────────────────────────
623
private emit(taskId: string, type: string, data: unknown): void {
  this.progressEmitter.emit(`task:${taskId}`, { type, data });
}
627
private async runTask(task: Task): Promise<void> {
  task.status = "running";
  this.emit(task.id, "started", { taskId: task.id, workflowType: task.workflowType });
  auditLogger.write({ event: "task.started", actorId: ACTOR_SYSTEM, taskId: task.id, data: { workflowType: task.workflowType } });
  const phases = PHASE_SEQUENCES[task.workflowType];
633
  for (const phase of phases) {
    if (task.currentRound >= task.maxRounds) {
      logger.warn("Task hit maxRounds cap", { taskId: task.id, maxRounds: task.maxRounds });
      break;
    }
    task.currentPhase = phase;
    task.updatedAt = new Date();
    this.emit(task.id, "phase", { phase });
    await this.runPhase(task, phase);
643
    // Wait for any pending gates before continuing
    if (task.pendingGates.some((g) => g.status === "pending")) {
      task.status = "awaiting_gate";
      await this.waitForGates(task);
      task.status = "running";
    }
  }
651
  // Final synthesis by root orchestrator
  task.output = await this.synthesise(task);
654
  // For tabulate workflows, also produce a structured spreadsheet-style table.
  if (task.workflowType === "tabulate") {
    try {
      task.table = await this.tabulate(task);
    } catch (err) {
      logger.warn("Tabulation failed; falling back to text output only", {
        taskId: task.id,
        error: (err as Error).message,
      });
    }
  }
666
  task.status = "complete";
  task.completedAt = new Date();
  task.updatedAt = new Date();
670
  // Close the time entry now that the task has finished.
  if (task.activeTimeEntryId) {
    const closedOnComplete = this.time.close(task.activeTimeEntryId);
    task.activeTimeEntryId = undefined;
    if (closedOnComplete) {
      jobQueue.enqueue("summarize_time_entry", {
        entryId: closedOnComplete.id,
        taskId: task.id,
        clientNumber: task.clientNumber,
      }).catch((e) => logger.warn("Failed to enqueue summarize job", { error: (e as Error).message }));
    }
  }
683
  // Feed agent performance back into the registry so recommend()-based
  // recruitment improves over time. High-confidence verified findings → positive
  // signal; challenged/low-confidence ones → negative signal.
  this.recordAgentOutcomes(task).catch((err) =>
    logger.warn("Agent outcome recording failed", { error: (err as Error).message }),
  );
690
  this.emit(task.id, "complete", { findings: task.findings.length, output: task.output?.slice(0, 200) });
  auditLogger.write({ event: "task.complete", actorId: ACTOR_SYSTEM, taskId: task.id, data: { findings: task.findings.length } });
  this.persistTasks().catch((err) => logger.warn("Failed to persist tasks", { error: err.message }));
694
  logger.info("Task complete", { taskId: task.id, findings: task.findings.length });
}
697
/**
 * Feed task outcomes back into the agent registry's performance scores.
 * Agents whose findings were verified and not challenged → high signal.
 * Agents whose findings were challenged, rejected, or very low confidence → low signal.
 * The registry writes these as `successScore` payload fields so future
 * recommend()-based recruitment can blend semantic match with proven performance.
 */
private async recordAgentOutcomes(task: Task): Promise<void> {
  if (!task.findings.length) return;
707
  // Resolve each finding's phase via a round→phase map (each round runs exactly
  // one phase) instead of an O(findings²) reverse-search through task.rounds.
  // The previous fallback to task.currentPhase always attributed unmatched
  // findings to the final ("delivery") phase, poisoning the per-phase Q-table.
  const phaseByRound = new Map<number, string>();
  for (const r of task.rounds) phaseByRound.set(r.goal.round, r.goal.phase);
714
  // Group findings by (agentId, phase) so Q-learning gets per-phase rewards.
  const agentPhaseScores = new Map<string, { phase: string; scores: number[] }>();
  for (const f of task.findings) {
    const phase = phaseByRound.get(f.round) ?? task.currentPhase;
    const key = `${f.agentId}::${phase}`;
    if (!agentPhaseScores.has(key)) {
      agentPhaseScores.set(key, { phase, scores: [] });
    }
    const effective = f.challenged && !f.resolved ? f.confidence * 0.3 : f.confidence;
    agentPhaseScores.get(key)!.scores.push(effective);
  }
726
  const phases = ["intake","research","analysis","drafting","review","verification","delivery"];
  const done = true; // task is complete
729
  for (const [key, { phase, scores }] of agentPhaseScores) {
    const agentId = key.split("::")[0];
    const avg = scores.reduce((a, b) => a + b, 0) / scores.length;
    const phaseIdx = phases.indexOf(phase);
    const nextPhase = phases[phaseIdx + 1] ?? phase;
735
    // Update Qdrant successScore (for recommend() in future tasks)
    await this.registry.recordOutcome([agentId], avg);
738
    // Update RuVector Q-table (for per-phase agent ranking in future tasks)
    await agentLearning.recordEpisode({
      phase,
      nextPhase,
      jurisdiction: task.jurisdiction,
      workflowType: task.workflowType,
      agentId,
      reward: avg,
      done,
    });
  }
}
751
private async runPhase(task: Task, phase: TaskPhase): Promise<void> {
  logger.info("Phase starting", { taskId: task.id, phase });
  auditLogger.write({ event: "phase.start", actorId: ACTOR_SYSTEM, taskId: task.id, data: { phase } });
755
  // Root orchestrator generates the round goal for this phase
  const goal = await this.generateRoundGoal(task, phase);
  goal.round = ++task.currentRound;
759
  // Look up tone profile and billing attribution from the task's primary lawyer
  const primaryProfileId = task.createdByProfileId ?? task.assignedLawyerIds?.[0];
  const primaryProfile = primaryProfileId ? this.profiles.get(primaryProfileId) : undefined;
  const lawyerTone = primaryProfile?.toneProfile;
764
  // Build agent billing context when billing is enabled
  const billingCtx: AgentBillingCtx | undefined = Config.agentBilling.enabled
    ? {
        timeStore: this.time,
        responsibleLawyerId: primaryProfileId,
        responsibleLawyerName: primaryProfile?.name,
        matterNumber: task.matterNumber,
        clientNumber: task.clientNumber,
      }
    : undefined;
775
  // Run DyTopo round
  const roundState = await this.engine.runRound(task, goal, lawyerTone, billingCtx);
  task.rounds.push(roundState);
779
  // Build source-text map for citation gate (from knowledge store)
  const sourceTexts = await this.buildSourceTextMap(task.documentIds);
782
  // Apply protocols to raw findings
  const rawFindings = roundState.findings;
  const { passed } = applyCitationGate(rawFindings, sourceTexts);
786
  // Debate each passing finding. allSettled so a single model error (429/500)
  // on one finding does not reject the whole phase and fail the entire task,
  // discarding every other finding. A failed debate leaves the finding as-is.
  const debateResults = await Promise.allSettled(
    passed.map((f) => runDebate(f, "adversarial-challenger", task.id)),
  );
  const debated = debateResults.map((r, i) => {
    if (r.status === "fulfilled") return r.value;
    logger.warn("Debate failed for finding — keeping it unchanged", { findingId: passed[i].id, error: (r.reason as Error)?.message });
    return passed[i];
  });
798
  // Verification pipeline — mutates each finding in place, attaching its
  // verificationResult (read downstream by identifyGateRequests). allSettled
  // for the same resilience reason; a failed verification simply leaves the
  // finding without a verificationResult.
  const verifyResults = await Promise.allSettled(
    debated.map((f) => runVerificationPipeline(f, task.id)),
  );
  for (let i = 0; i < verifyResults.length; i++) {
    const r = verifyResults[i];
    if (r.status === "rejected") {
      logger.warn("Verification failed for finding", { findingId: debated[i].id, error: (r.reason as Error)?.message });
    }
  }
812
  // Add findings to task
  task.findings.push(...debated);
815
  // Identify gate requests
  const gates = identifyGateRequests(task.id, debated);
  task.pendingGates.push(...gates);
819
  task.updatedAt = new Date();
  this.emit(task.id, "round", {
    round: task.currentRound,
    phase,
    findings: debated.length,
    gates: gates.length,
  });
  auditLogger.write({ event: "phase.complete", actorId: ACTOR_SYSTEM, taskId: task.id, data: { phase, findings: debated.length, gates: gates.length } });
  logger.info("Phase complete", {
    taskId: task.id,
    phase,
    findings: debated.length,
    gates: gates.length,
  });
}
835
private async generateRoundGoal(task: Task, phase: TaskPhase): Promise<RoundGoal> {
  const safeDesc = sanitizePromptContent(task.description);
  const priorPhases = task.rounds.map((r) => r.goal.phase);
  const prompt = `TASK: ${safeDesc}
840
841WORKFLOW: ${task.workflowType}
842CURRENT PHASE: ${phase}
843PRIOR PHASES COMPLETED: ${priorPhases.join(", ") || "none"}
844FINDINGS SO FAR: ${task.findings.length}
845
846Generate a specific, actionable round goal for the ${phase} phase.
847Format:
848DESCRIPTION: <one paragraph describing what agents should do this round>
849EXPECTED_OUTPUT_1: <first expected output>
850EXPECTED_OUTPUT_2: <second expected output>
851EXPECTED_OUTPUT_3: <third expected output>`;
852
  const model = selectModel({ tier: 0, taskType: "synthesis" });
  const provider = getProvider(model);
  const response = await provider.chat({
    model: resolveModelId(model),
    maxTokens: 600,
    system: ROOT_ORCHESTRATOR.systemPrompt,
    messages: [{ role: "user", content: prompt }],
    cacheSystem: true,
  });
862
  recordOrchestratorCost(response, model, "round_goal", task.id);
  const textBlock = response.content.find((b) => b.type === "text");
  const text = textBlock?.type === "text" ? textBlock.text : "";
  const descMatch = text.match(/DESCRIPTION:\s*([\s\S]+?)(?=EXPECTED_OUTPUT|$)/i);
  const outputMatches = [...text.matchAll(/EXPECTED_OUTPUT_\d+:\s*(.+)/gi)];
868
  return {
    id: uuidv4(),
    round: task.currentRound,
    phase,
    description: descMatch?.[1]?.trim() ?? `Execute the ${phase} phase for: ${safeDesc}`,
    expectedOutputs: outputMatches.map((m) => m[1].trim()),
  };
}
877
private async synthesise(task: Task): Promise<string> {
  const safeDesc = sanitizePromptContent(task.description);
  // Cap each finding and the total to prevent synthesis prompts from exceeding
  // practical context-window limits when many rounds produce large findings.
  const findingsSummary = task.findings
    .filter((f) => !task.pendingGates.some((g) => g.findingId === f.id && g.status === "rejected"))
    .map((f, i) => `[${i + 1}] (${f.agentName}, Round ${f.round}) ${f.content.slice(0, 5_000)}`)
    .join("\n\n")
    .slice(0, 200_000);
887
  const lawyerTone = (() => {
    const profileId = task.createdByProfileId ?? task.assignedLawyerIds?.[0];
    if (!profileId) return undefined;
    return this.profiles.get(profileId)?.toneProfile;
  })();
893
  const toneBlock = lawyerTone
    ? `\nLAWYER TONE PROFILE — write the final output in this voice:\n${sanitizePromptContent(lawyerTone.injectionSnippet)}\n`
    : "";
897
  const prompt = `TASK: ${safeDesc}
899
900ALL FINDINGS FROM ALL ROUNDS:
901${findingsSummary}
902${toneBlock}
903Produce the final legal output for this task. Structure appropriately for the workflow type: ${task.workflowType}.
904Every claim must trace to a specific finding number from the list above.`;
905
  const model = selectModel({ tier: 0, taskType: "synthesis" });
  const useThinking = shouldUseThinking({ modelId: model, taskType: "synthesis", tier: 0 });
  const provider = getProvider(model);
  const response = await provider.chat({
    model: resolveModelId(model),
    // When thinking is on, budget_tokens + output tokens must fit within max_tokens.
    maxTokens: useThinking ? 16_000 : 4000,
    system: ROOT_ORCHESTRATOR.systemPrompt,
    messages: [{ role: "user", content: prompt }],
    cacheSystem: true,
    ...(useThinking && { thinking: { budgetTokens: Config.anthropic.thinkingBudgetTokens } }),
  });
918
  recordOrchestratorCost(response, model, "synthesis", task.id);
  const textBlock = response.content.find((b) => b.type === "text");
  return textBlock?.type === "text" ? textBlock.text : "";
}
923
/**
 * Extract the task's findings into a structured table for spreadsheet-style
 * review. The root orchestrator chooses appropriate columns for the subject
 * matter and maps each row back to a source finding via `_findingId`.
 */
private async tabulate(task: Task): Promise<TaskTable | undefined> {
  const safeDesc = sanitizePromptContent(task.description);
  const findings = task.findings.filter(
    (f) => !task.pendingGates.some((g) => g.findingId === f.id && g.status === "rejected"),
  );
  if (findings.length === 0) return undefined;
935
  const findingsSummary = findings
    .map((f) => `id=${f.id} | ${f.agentName} (R${f.round}, conf ${f.confidence.toFixed(2)}): ${f.content}`)
    .join("\n\n");
939
  const prompt = `TASK: ${safeDesc}
941
942FINDINGS:
943${findingsSummary}
944
945Extract these findings into a structured table suitable for a spreadsheet review grid.
946If the TASK above names the columns it wants, use exactly those column names and order.
947Otherwise choose 3–6 columns that best capture the structured content for THIS subject matter.
948
949Respond with ONLY valid JSON (no prose, no markdown fences) in exactly this shape:
950{
"columns": ["Column A", "Column B"],
"rows": [
  { "Column A": "value", "Column B": "value", "_findingIds": ["<source finding id>", "..."] }
]
955}
956
957Rules:
958- MERGE findings that make the same substantive point into a SINGLE row.
959- Keep findings that make genuinely DIFFERENT points as SEPARATE rows, even when they concern the same clause or topic.
960- "_findingIds" MUST list the bare id(s) of every finding that contributes to that row (no "id=" prefix or brackets).
961- Do NOT add columns for source ids or confidence — those are attached automatically.
962- Do NOT use column names beginning with an underscore.
963- Every column listed in "columns" must appear as a key in every row.
964- Keep cell values concise (a phrase or sentence, not paragraphs).`;
965
  const model = selectModel({ tier: 0, taskType: "synthesis" });
  const provider = getProvider(model);
  const response = await provider.chat({
    model: resolveModelId(model),
    maxTokens: 4000,
    system: ROOT_ORCHESTRATOR.systemPrompt,
    messages: [{ role: "user", content: prompt }],
    cacheSystem: true,
  });
975
  recordOrchestratorCost(response, model, "tabulate", task.id);
  const textBlock = response.content.find((b) => b.type === "text");
  const text = textBlock?.type === "text" ? textBlock.text : "";
979
  const parsed = parseJsonObject(text) as
    | { columns?: unknown; rows?: unknown }
    | undefined;
  if (!parsed || !Array.isArray(parsed.columns) || !Array.isArray(parsed.rows)) {
    logger.warn("Tabulation produced unparseable output", { taskId: task.id });
    return undefined;
  }
987
  const confByFinding = new Map(findings.map((f) => [f.id, f.confidence]));
  const validIds = new Set(findings.map((f) => f.id));
  // Display columns never include underscore-prefixed internal keys.
  const columns = parsed.columns.map((c) => String(c)).filter((c) => !c.startsWith("_"));
992
  const UUID = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi;
  // Collect every known finding id referenced anywhere in the row (the model
  // may use `_findingIds`, `_findingId`, or embed ids inside a stray value).
  const contributorsOf = (r: Record<string, unknown>): string[] => {
    const found = new Set<string>();
    const pool: string[] = [];
    const raw = (r._findingIds ?? r._findingId) as unknown;
    if (Array.isArray(raw)) pool.push(...raw.map(String));
    else if (raw != null) pool.push(String(raw));
    for (const v of Object.values(r)) pool.push(String(v ?? ""));
    for (const s of pool) for (const m of s.matchAll(UUID)) if (validIds.has(m[0])) found.add(m[0]);
    return [...found];
  };
1006
  const byConfDesc = (a: string, b: string) => (confByFinding.get(b) ?? 0) - (confByFinding.get(a) ?? 0);
1008
  const rawRows = (parsed.rows as Array<Record<string, unknown>>)
    .filter((r) => r && typeof r === "object")
    .map((r) => {
      const row: Record<string, string> = {};
      for (const col of columns) row[col] = r[col] != null ? String(r[col]) : "";
      const contributors = contributorsOf(r).sort(byConfDesc);
      const confs = contributors.map((id) => confByFinding.get(id)).filter((c): c is number => c != null);
      row._findingIds = contributors.join(",");
      row._findingId = contributors[0] ?? "";
      row._confidence = confs.length ? Math.max(...confs).toFixed(2) : "";
      row._sources = String(contributors.length);
      return row;
    });
1022
  // Safety net: collapse rows whose visible cells are identical (the model
  // missed a merge). Genuinely distinct points differ in their cells and are
  // preserved — each keeps its own confidence.
  const merged = new Map<string, Record<string, string>>();
  for (const row of rawRows) {
    const key = columns.map((c) => (row[c] ?? "").trim().toLowerCase()).join("");
    const existing = merged.get(key);
    if (!existing) { merged.set(key, { ...row }); continue; }
    const ids = [...new Set([
      ...existing._findingIds.split(",").filter(Boolean),
      ...row._findingIds.split(",").filter(Boolean),
    ])].sort(byConfDesc);
    existing._findingIds = ids.join(",");
    existing._findingId = ids[0] ?? existing._findingId;
    existing._sources = String(ids.length);
    const maxConf = Math.max(parseFloat(existing._confidence || "0"), parseFloat(row._confidence || "0"));
    existing._confidence = maxConf ? maxConf.toFixed(2) : "";
  }
1041
  const rows = [...merged.values()].sort(
    (a, b) => parseFloat(b._confidence || "0") - parseFloat(a._confidence || "0"),
  );
  const sourceFindingIds = [...new Set(rows.flatMap((r) => r._findingIds.split(",").filter(Boolean)))];
1046
  return { columns, rows, sourceFindingIds, generatedAt: new Date() };
}
1049
private async buildSourceTextMap(docIds: string[]): Promise<Map<string, string>> {
  const map = new Map<string, string>();
  await Promise.all(
    docIds.map(async (id) => {
      const text = await this.knowledge.getFullText(id);
      if (text) map.set(id, text);
    }),
  );
  return map;
}
1060
// ─── Persistence ──────────────────────────────────────────────────────────
1062
async persistTasks(): Promise<void> {
  const path = Config.persistence.tasksFile;
  const serialisable = Array.from(this.tasks.values()).map((t) => ({
    ...t,
    createdAt: t.createdAt.toISOString(),
    updatedAt: t.updatedAt.toISOString(),
    completedAt: t.completedAt?.toISOString(),
  }));
  const tmp = `${path}.tmp`;
  await writeFile(tmp, JSON.stringify(serialisable, null, 2), "utf8");
  await rename(tmp, path);
  logger.debug("Tasks persisted", { count: this.tasks.size, path });
}
1076
async restoreTasks(): Promise<void> {
  const path = Config.persistence.tasksFile;
  let raw: string;
  try {
    raw = await readFile(path, "utf8");
  } catch {
    return; // no file yet — first run
  }
1085
  try {
    const items = JSON.parse(raw) as Array<Record<string, unknown>>;
    for (const item of items) {
      // Validate workflowType before restoring — a tampered or corrupted file
      // could produce an invalid type that crashes phase execution at runTask().
      if (!PHASE_SEQUENCES[item.workflowType as WorkflowType]) {
        logger.warn("Skipping restored task with unknown workflowType", { id: item.id, workflowType: item.workflowType });
        continue;
      }
      const task = {
        ...item,
        createdAt: new Date(item.createdAt as string),
        updatedAt: new Date(item.updatedAt as string),
        completedAt: item.completedAt ? new Date(item.completedAt as string) : undefined,
      } as Task;
      this.tasks.set(task.id, task);
    }
    for (const task of this.tasks.values()) {
      if (task.status === "running" || task.status === "pending" || task.status === "awaiting_gate") {
        task.status = "failed";
        task.error = "Server restarted during execution";
      }
    }
    logger.info("Tasks restored from disk", { count: this.tasks.size, path });
  } catch (err) {
    logger.warn("Failed to restore tasks", { error: (err as Error).message });
  }
}
1114
private waitForGates(task: Task): Promise<void> {
  return new Promise((resolve) => {
    if (task.pendingGates.every((g) => g.status !== "pending")) {
      resolve();
      return;
    }
    const GATE_TIMEOUT_MS = 72 * 60 * 60 * 1000;
    const timer = setTimeout(() => {
      this.gateEmitter.off(`gates:${task.id}`, handler);
      logger.warn("Gate timeout — auto-resolving", { taskId: task.id });
      resolve();
    }, GATE_TIMEOUT_MS);
    const handler = () => {
      if (task.pendingGates.every((g) => g.status !== "pending")) {
        clearTimeout(timer);
        this.gateEmitter.off(`gates:${task.id}`, handler);
        resolve();
      }
    };
    this.gateEmitter.on(`gates:${task.id}`, handler);
  });
}
1137}
No results