Forem: Jason

Markus: An Open-Source AI Digital Workforce Platform with Organizational Governance

Jason — Tue, 05 May 2026 15:39:54 +0000

Abstract

Large language model (LLM) agents are rapidly transitioning from research prototypes to production deployments, yet existing agent frameworks remain fundamentally oriented toward individual task completion rather than organizational collaboration. Current platforms—including LangChain, AutoGen, and CrewAI—provide sophisticated tools for single-agent reasoning and multi-agent conversation, but lack the organizational infrastructure required for agents to function as collaborative team members: structured task lifecycles with review and approval workflows, hierarchical memory systems that persist knowledge across sessions, attention control mechanisms that manage competing communication demands, and governance policies that ensure accountability and auditability.

This paper presents Markus, an open-source AI Digital Workforce Platform that treats agents as organizational members with full team coordination capabilities. Markus introduces seven core technical contributions: (1) an organizational agent architecture with role-based identity management and workspace isolation; (2) a hierarchical task management system with DAG dependency scheduling and lifecycle state machines; (3) a five-layer memory system inspired by the Baddeley working memory model, with explicit promotion mechanisms and file-first persistence; (4) a mailbox-based attention controller that provides priority-ordered, interrupt-safe context switching; (5) workspace isolation with mutual review protocols and A2A communication; (6) a Cognitive Preparation Pipeline (CPP) grounded in Kahneman's dual-process theory that performs multi-phase context assembly before each LLM call; and (7) a dual-mode skill system supporting both instruction-based and MCP-based extensions. Markus is open-source and production-deployed, managing over 20 active agents executing hundreds of tasks across multiple teams. Preliminary evaluation from controlled pilot studies demonstrates a 2.3× improvement in multi-agent task throughput compared to manual orchestration, 47% better recall over long sessions versus flat context baselines, and 35% reduction in context token waste via the Cognitive Preparation Pipeline. These results are initial estimates based on pilot experiments and should be interpreted as indicative rather than conclusive.

1. Introduction

Context. The emergence of large language models (LLMs) has catalyzed a paradigm shift in how software systems interact with humans and execute complex tasks. LLM-based agents—autonomous programs that leverage LLMs for reasoning, planning, and tool use—are increasingly deployed in production environments for code generation, data analysis, customer support, and workflow automation [Wang et al., 2024; Xi et al., 2023]. As organizations scale their adoption of AI agents, a critical gap has become apparent: current frameworks are designed for individual capability, not organizational collaboration.

The Three Core Challenges. We identify three fundamental challenges that existing platforms fail to address:

C1: The Single-Agent Ceiling. Most agent frameworks (LangChain [Chase, 2022], LlamaIndex [Liu, 2023], Semantic Kernel [Microsoft, 2023]) focus on equipping a single agent with tools, memory, and reasoning capabilities. Even when multi-agent support is added, collaboration remains at the level of simple message passing or function calls—there is no governance layer for approval, review, delegation, or escalation.

C2: The Passive Interaction Model. Contemporary agents operate on a request-response paradigm: they wait for a user query, process it, and return a result. They cannot proactively work through a task backlog, periodically report progress, escalate blockers, or initiate knowledge sharing. This fundamentally limits their utility as autonomous team members.

C3: The Missing Organizational Layer. Real-world software teams operate within structured workflows: requirements are specified, tasks are assigned and prioritized, work products undergo peer review, and lessons learned are documented as reusable knowledge. Existing agent platforms provide none of these organizational primitives. Agents operate in isolation, without role definitions, project boundaries, audit trails, or quality gates.

Contribution. This paper presents Markus, an open-source AI Digital Workforce Platform that addresses these challenges by embedding agents within a complete organizational framework. Markus reimagines agents not as stateless function-callers but as digital employees with identity, roles, tasks, memory, and accountability. The platform implements seven key innovations:

Organizational Agent Architecture (Section 3.1–3.2): Agents are organized into teams under projects, with role-based identity files (ROLE.md), workspace isolation, and single-threaded attention models that prevent cognitive interference.
Hierarchical Task Management (Section 3.3): A structured task system with unified lifecycle state machines, DAG-based dependency scheduling, mandatory peer review, and configurable governance policies.
Five-Layer Memory System (Section 3.4): A cognitively-grounded memory hierarchy spanning from identity (L1) through working context (L5), with explicit promotion mechanisms and dream cycles for automatic knowledge consolidation.
Mailbox-Based Attention Control (Section 3.5): An asynchronous messaging system with priority queues, urgency-aware scheduling, and selective interruption that manages "who gets to communicate with the agent and when."
Workspace Isolation and Team Collaboration (Section 3.6): Agents work in separated workspaces with git worktree isolation, mutual review protocols, and A2A communication via typed message modes.
Cognitive Preparation Pipeline (Section 3.7): A multi-phase pre-processing pipeline—inspired by Kahneman's dual-process theory [Kahneman, 2011]—that performs appraisal, directed retrieval, reflection, and context assembly before each LLM call, reducing noise and improving response quality.
Dual-Mode Skill System (Section 3.8): A skill architecture supporting both lightweight instruction-based extensions (SKILL.md) and full MCP-based server plugins.

Markus is released as open-source software under AGPL-3.0. It is deployed in production environments managing 20+ agents across multiple teams, executing hundreds of tasks with full audit trails. The platform has received contributions from the open-source community and has been adopted for use cases spanning software development, research analysis, content creation, and quantitative finance.

The remainder of this paper is organized as follows. Section 2 discusses related work in agent frameworks, multi-agent systems, memory architectures, and organizational models. Section 3 details Markus's system design and architecture. Section 4 presents our evaluation methodology and results. Section 5 discusses design tradeoffs, lessons learned, and limitations. Section 6 concludes and outlines future work.

2. Related Work

2.1 LLM Agent Frameworks

The rapid advancement of LLMs has spawned numerous agent frameworks. LangChain [Chase, 2022] provides a modular toolkit for chaining LLM calls with tools, retrievers, and memory—establishing many of the foundational patterns for LLM application development. LlamaIndex [Liu, 2023] specializes in data indexing and retrieval-augmented generation (RAG). Semantic Kernel [Microsoft, 2023] offers enterprise-grade LLM orchestration with planner capabilities. While these frameworks excel at single-agent task composition, they provide minimal support for multi-agent coordination. Agents built on these platforms operate independently; there is no shared task board, no cross-agent knowledge sharing, and no organizational governance. Markus differs by placing agents within an organizational hierarchy from the outset, providing built-in structures for team coordination.

2.2 Multi-Agent Systems

Several platforms have extended agent frameworks to support multiple agents. AutoGen [Wu et al., 2023] from Microsoft Research enables multi-agent conversations with role-based delegation and tool use, introducing the concept of agent-to-agent message passing. CrewAI [João, 2024] provides a role-based multi-agent framework with simple task assignment and sequential process flows. MetaGPT [Hong et al., 2023] simulates a software company with specialized agent roles (product manager, architect, engineer) that collaborate through structured document exchange. ChatDev [Qian et al., 2023] uses a similar role-playing approach for software development.

A fundamental limitation shared by these platforms is the absence of structured task lifecycles. Agents in AutoGen and CrewAI collaborate through conversation—there is no concept of task states (pending, in_progress, review, completed), no mandatory peer review gate, no dependency-aware scheduling, and no governance policies (concurrency limits, approval tiers, deletion protection). Markus introduces a full task lifecycle state machine with review-mandatory workflows, DAG-based task dependencies, and configurable project governance. Furthermore, Markus's workspace isolation (via git worktree) and mutual review protocol ensure that agents in a team do not interfere with each other's work—a concern not addressed by existing multi-agent platforms.

2.3 Memory Systems for AI Agents

Memory in AI agents has received growing attention. MemGPT [Packer et al., 2024] introduces virtual context management with a two-tier memory system (working context + external storage) that manages context windows by swapping information in and out. LangMem and Mem0 provide persistent external memory for LLM applications using vector databases.

Markus's five-layer memory system differs in several important ways. First, it is explicitly grounded in cognitive psychology—citing Baddeley's working memory model [Baddeley, 2000] and distinguishing between short-term, episodic, and long-term storage. Second, it adopts a file-first persistence strategy: agent memories are stored as human-readable Markdown files, not opaque database entries, enabling manual inspection, editing, and portability. Third, the system implements an explicit promotion mechanism: observations from the buffer (L3) are promoted to curated knowledge (L2) only after validation, and recurring patterns (3+ occurrences) are automatically consolidated via dream cycles. Fourth, the memory system integrates with the Cognitive Preparation Pipeline for targeted retrieval rather than bulk loading, maximizing signal-to-noise ratio in the limited context window [Markus, 2026a].

2.4 Organizational and Cognitive Models

Beyond agent-specific systems, Markus draws inspiration from workflow engines and cognitive architectures. Temporal [Temporal, 2024] and Airflow [Apache, 2024] provide DAG-based workflow orchestration for distributed systems, but are not designed for agent-aware task execution—they cannot route tasks to agents based on skills or hold agents accountable through review workflows. ACT-R [Anderson et al., 2004] and SOAR [Laird, 2012] are cognitive architectures that model human problem-solving with structured memory modules and production rules, but are research frameworks not designed for production LLM agent deployment.

Markus occupies a unique intersection: it applies organizational theory from software engineering (code review, CI/CD, project governance) to the design of multi-agent systems, while incorporating cognitive architecture principles (dual-process theory, working memory model) into the agent execution pipeline. This synthesis—organizational governance plus cognitive architecture—is, to our knowledge, novel in the literature.

3. System Design and Architecture

3.1 System Overview

Markus adopts a layered architecture that separates presentation, API, core services, and storage into independent tiers. The top layer is a React-based Web UI providing dashboards for chat, agent management, task boards, team coordination, and governance configuration. The API Server (Node.js) exposes REST endpoints and WebSocket connections for real-time communication. The core service layer comprises six primary services: Organization Management (OrgSvc), Task Management (TaskSvc), Agent Manager (AgentMgr), Project Management (ProjectSvc), Knowledge Management (KnowledgeSvc), and the Governance Layer. The storage layer uses SQLite for structured data (tasks, agents, organization relationships), the filesystem for agent memories and sessions, and a vector store for semantic retrieval.

A distinguishing architectural choice is Markus's file-system centric design: there are no external database dependencies. SQLite is embedded, vector storage is optional, and all agent-specific cognitive data (identity, memory, sessions, skills) is stored as files within ~/.markus/agents/{agentId}/. This design reduces operational complexity and ensures that agent data remains portable and human-accessible.

3.2 Agent Runtime: Single-Threaded Cognitive Entities

Contribution Statement 1: Markus models each agent as a single-threaded cognitive entity—an explicit design departure from the concurrent, multi-threaded approaches common in existing frameworks.

Each Markus agent processes one task at a time, serializing all external stimuli through a priority mailbox. This design decision emerged from early experiments with concurrent message handling, which revealed three persistent problems:

Memory contamination: Concurrent sessions polluted each other's context windows
Cognitive interference: An agent writing code would be interrupted mid-thought by a chat message, degrading output quality
Nondeterministic behavior: Race conditions on state changes made debugging nearly impossible

The single-threaded model trades theoretical peak throughput for predictability, debuggability, and output quality. An agent's lifecycle includes seven states: creation and onboarding (identity file generation), startup and registration (EventBus and WebSocket connection), periodic heartbeat (every ~30 minutes for self-checks), active work and idle waiting, pause and resume for maintenance, and graceful shutdown.

Each agent's identity is defined by files in its private workspace:

ROLE.md: The agent's complete identity—who it is, what it does, how it works. Always loaded as a stable system prompt prefix. Maximum 200 lines.
HEARTBEAT.md: Self-check checklist loaded during each heartbeat cycle.
MEMORY.md: Curated long-term knowledge. Always loaded as the "Your Knowledge" section.
POLICIES.md: Safety constraints and behavioral boundaries. Loaded on demand.
CONTEXT.md: Domain-specific context and references. Loaded on demand.

The AgentManager serves as the central controller, managing the lifecycle of all agent processes, maintaining a registry with EventBus mappings, and forwarding agent events to the WebSocket broadcast layer for UI updates.

3.3 Hierarchical Task Management with DAG Dependencies

Contribution Statement 2: Markus introduces a hierarchical task management system with a unified lifecycle state machine, DAG-based dependency scheduling, and mandatory review workflows—applying software engineering governance practices to AI agent task execution.

Markus organizes work into three hierarchical levels:

Requirement ── describes "what" and "why" (user-authorized)
    └── Task ── discrete, executable work unit (with assignee + reviewer)
         └── Subtask ── trackable checklist item (within a task)

A core design decision is the Unified Status Vocabulary: tasks and requirements share the same state enumeration, ensuring that the same state name always conveys the same meaning across contexts:

Status	Meaning	Notes
`pending`	Awaiting approval	Created but not yet authorized
`in_progress`	Being executed	Approved, actively working
`blocked`	Waiting on dependency	E.g., waiting for upstream task
`review`	Under review	Execution done, reviewer evaluating
`completed`	Successfully finished	Per reviewer approval
`failed`	Unrecoverable error	Retry policy may apply
`rejected`	Proposal denied	Never started
`cancelled`	Stopped mid-work	Intentional halt
`archived`	Historical record	No longer active

The task state machine enforces Markus's core workflow philosophy:

Approval-first: Tasks enter pending and require human or manager approval before execution
Review-mandatory: Execution triggers automatic transition to review; only reviewer approval completes the task
Revision loop: Rejected review sends the task back to in_progress for another execution pass
Auto-retry: Failed tasks can be configured with retry strategies that restart execution automatically

DAG-based task scheduling: Tasks express dependencies through a blocked_by field containing task IDs. The system automatically constructs a directed acyclic graph (DAG) of tasks—independent tasks execute in parallel, while dependent tasks await their predecessors. When a predecessor completes, downstream tasks are automatically unblocked. This mechanism enables complex multi-agent workflows to be orchestrated through simple dependency declarations, without requiring manual coordination.

Supporting capabilities include scheduled (cron-based) tasks with the same review lifecycle, configurable governance policies (approval tiers, concurrency limits per agent, git command authorization), and data retention rules (cancelled/failed tasks retained for 7 days, completed tasks archived after 30 days).

3.4 Five-Layer Memory System

Contribution Statement 3: Markus's five-layer memory system, inspired by Baddeley's working memory model [Baddeley, 2000], provides a structured hierarchy for information retention with explicit promotion mechanisms and file-first persistence.

The memory system is organized into five layers with increasing persistence and decreasing volatility:

Layer	Storage	Injection	Capacity	Purpose
L1: Identity	`role/ROLE.md`	Always loaded	200 lines max	Agent identity definition
L2: Knowledge	`MEMORY.md`	Always loaded	15000 chars (3000/section)	Validated long-term knowledge
L3: Observations	`memories.json`	CPP-guided retrieval	50+ triggers dream cycle	Raw observations, insights
L4: Experience	SQLite `agent_activities`	CPP Phase 2 retrieval	Unbounded	Execution history
L5: Working Context	`sessions/*.json`	Message history	80->40 auto-compressed	Current session

Design principles: Five principles guide the memory system.

Layered Retention: Information migrates from volatile short-term context toward persistent long-term knowledge through explicit promotion mechanisms—never automatic. Agents use memory_save for buffer entries and memory_update_longterm for curated knowledge.

File-First: The primary storage medium is local files (~/.markus/agents/{id}/). Files are human-readable Markdown, enabling manual inspection, editing, and migration. SQLite handles structured activity data; the filesystem handles cognitive data.

Context as Currency: Every byte injected into the LLM context competes for limited context window space. The system avoids "throw all memories into context" approaches, relying instead on the Cognitive Preparation Pipeline for targeted, high-signal retrieval.

Agent Autonomy: The agent decides what to remember, what to refine, and how to evolve. The system provides mechanisms; the agent drives strategy through skill-guided behavior.

Cognitive Retrieval: Memory access is role-directed, not batch-loaded. The CPP retrieves based on the agent's role, state, and current scenario.

The Observation Buffer and Dream Cycle: The observation buffer (memories.json) is the most active memory layer. Agents write raw observations, tool tips, and task outcomes via memory_save, each tagged with type (fact, note, insight) and custom tags for classification. When entries exceed 50, a dream cycle triggers: an LLM-assisted consolidation process that identifies recurring patterns (3+ occurrences), promotes them to MEMORY.md, and prunes the source entries. The dream cycle is conservative—false retention is preferred over false deletion.

3.5 Mailbox-Based Attention Control

Contribution Statement 4: Markus introduces a mailbox-based attention controller that serializes all agent-directed stimuli through a priority queue with urgency-aware scheduling, solving the fundamental problem of "who gets to communicate with the agent and when."

Each agent owns a priority mailbox that processes incoming messages in order of priority (critical > high > normal > low > background). Fourteen predefined source types are categorized into four classes:

Category	Types	Description
Interaction	`human_chat`, `a2a_message`, `mention`	Direct dialog communication
Task	`task_status_update`, `task_comment`, `review_request`	Task lifecycle events
Notification	`requirement_update`	Status change notifications
System	`system_event`, `heartbeat`, `daily_report`	Internal agent processes

Execution-mode task messages (with triggerExecution=true) are assigned high priority for timely processing. The Attention Controller—an event-driven focus manager—handles interrupts only at safe yield points (e.g., after completing a tool call, not mid-response generation), preventing agents from being disrupted during critical operations. All attention decisions are logged to the agent_decisions table, creating a complete attention audit trail.

Notable mailbox processing rules include: task_status_update supports two modes—execution mode, which triggers a full executeTask() loop with LLM invocation, and information mode, which auto-acknowledges without LLM cost; heartbeat messages load HEARTBEAT.md and run the self-check checklist; daily_report messages trigger automated report generation with user feedback ingestion.

3.6 Team Collaboration and Workspace Isolation

Contribution Statement 5: Markus implements a three-tier organizational hierarchy (Organization, Team, Agent) with workspace isolation, mutual review protocols, and trust-based autonomy adjustment.

Markus adopts a three-layer organizational model:

Organization
 └── Teams
      ├── Manager (human or agent)
      └── Members (agents and humans mixed)
           └── Projects (work scope with governance)

The team collaboration workflow consists of five phases:

Plan: The manager decomposes requirements into a task DAG, specifying dependencies and execution order
Execute: Members use git worktree to isolate their working branches, executing tasks in parallel when independent or sequentially when dependent
Review & Merge: Completed tasks enter review; the reviewer inspects deliverables, approves or requests changes. On approval, code is merged
Validate: Integration tests and CI checks run on the merged code
Consolidate: Lessons learned are registered as team deliverables for future reuse

Agent-to-Agent (A2A) Communication uses a distinct tool (agent_send_message) with two modes:

Notification mode (waitForReply=false): One-way broadcast for status updates ("I submitted task X for review")
Q&A mode (waitForReply=true): Blocking request for feedback or decisions

A critical design constraint: substantial cross-agent work requests must go through the requirement + task system, not informal messaging. This ensures all work is tracked, reviewed, and auditable.

Workspace isolation prevents agent interference: each agent owns a dedicated workspace (~/.markus/agents/{agentId}/), file write tools are scoped to the agent's own directory and a shared directory, and task execution uses git worktree for branch-level isolation. The reviewer verifies that changes stay within task boundaries before approving.

Trust scoring provides dynamic autonomy adjustment based on delivery quality:

Successful first-pass deliveries increase trust
Revisions and rejections decrease trust
Trust levels (Probation, Standard, Trusted, Senior) determine autonomy—how many tasks can auto-approve and whether the agent can review others' work

3.7 Cognitive Preparation Pipeline

Contribution Statement 6: Markus's Cognitive Preparation Pipeline (CPP) is an engineering implementation of Kahneman's dual-process theory [Kahneman, 2011], performing System-2-style context preparation before each primary LLM call.

Traditional agents operate in System 1 mode: upon receiving a stimulus, they mechanically assemble context and respond. CPP introduces a System 2 phase—a preparatory step where the agent "thinks about what it needs to know" before the main reasoning call. The pipeline has five phases:

Phase 1: Appraisal — Agent assesses the situation and plans needed retrieval
    ↓  (Cheapest model, 512 token output, no tools)
Phase 2: Directed Retrieval — Precise retrieval from memory, activity history, project knowledge
    ↓  (Code logic, no LLM call)
Phase 3: Reflection — Extract role-relevant patterns from retrieved context
    ↓  (Cheapest model, 512 token output, no tools)
Phase 4: Assembly — Assemble reflection results into PreparedContext, inject into system prompt
    ↓  (Code logic, no LLM call)
Phase 5: Evaluation — Deep mode only (D3): post-response assessment of context sufficiency
    ↓  (Cheapest model, 256 token output)

Three cognitive depth levels determine when CPP is engaged:

Level	Name	Use Case	CPP Phases
D1	Shallow	Heartbeat, simple queries	None—basic assembly only
D2	Standard	Regular task execution, interaction	Phases 1–4
D3	Deep	Complex problem-solving, strategic planning	Phases 1–5 + post-response evaluation

CPP produces measurable improvements: by pre-filtering and refining context before the main LLM call, it reduces token waste (fewer irrelevant memories injected), improves response relevance (targeted retrieval based on situational appraisal), and lowers cost (preparation uses cheap models, only the main call uses the full model).

3.8 Dual-Mode Skill System

Contribution Statement 7: Markus's dual-mode skill system supports both instruction-based and MCP-based extensions, balancing ease of development with tool-rich capability.

Skills are Markus's mechanism for extending agent capabilities. The system supports two modes:

Instruction-based skills are the simpler model: a SKILL.md file containing instructions that are injected into the agent's context, teaching the agent how to use existing tools in new ways. No new tools are added—the skill is pure instruction. Examples include the github skill (teaches gh CLI usage), the weather skill (teaches curl-based API queries), and the humanizer skill (teaches detection and removal of AI writing patterns).

MCP-based skills bundle a skill directory that may include scripts, configuration files, and resources. On installation, the entire directory is deployed to ~/.markus/skills/{name}/, the SKILL.md instructions are injected into the agent context, and any declared MCP servers are started to provide new tools. The manifest declares MCP server configurations with commands, arguments, and environment variables.

A notable built-in skill is self-evolution, which enables agents to continuously improve their own behavior:

Observation: Errors, corrections, and best practices are logged to the observation buffer
Promotion: Validated patterns are refined into MEMORY.md knowledge
Role evolution: Proven behavioral rules propagate to ROLE.md
Skill crystallization: Reusable team practices are packaged as installable skills

This self-evolution capability means Markus agents are not static programs—they improve with use, adapting their behavior based on task outcomes and user feedback.

3.9 Built-in Tool Infrastructure and Provider Routing

Tool System. Beyond the skill extension mechanism, Markus provides every agent with a comprehensive set of built-in tools organized into functional categories:

Category	Tools	Purpose
File System	`file_read`, `file_write`, `file_edit`, `grep_search`, `glob_find`, `list_directory`	Code and data file manipulation
Communication	`agent_send_message` (A2A), `notify_user`, `request_user_approval`	Inter-agent and human interaction
Task Management	`task_create`, `task_list`, `task_update`, `task_get`, `task_note`, `task_submit_review`	Full task lifecycle operations
Knowledge	`memory_save`, `memory_search`, `memory_list`, `memory_update_longterm`, `deliverable_create`, `deliverable_search`	Memory and knowledge management
Sub-Agent (Lightweight)	`spawn_subagent`, `spawn_subagents`	Delegate focused subtasks without creating new team agents
Information Retrieval	`web_fetch`, `web_search`	Real-time web data access

All shell_execute calls that involve Git operations automatically inject agent identity metadata via --author and --trailer flags (agent ID, team, organization, task ID), establishing a complete audit trail for every code change.

Provider Routing and Fault Tolerance. Markus's LLM provider routing layer implements production-grade high availability:

Provider-agnostic model catalog: Each provider maintains a model directory with capabilities (tool calling, streaming support), context window sizes, and default parameters.
Automatic failover: When the primary provider fails (network error, authentication failure, rate limiting), the routing layer automatically attempts backup providers.
Circuit breaker: After two consecutive failures, a provider is marked degraded for 5 minutes; authentication or billing errors trigger a 30-minute cooldown. Backup providers are used during this period.
Exponential backoff: withNetworkRetry() wraps all LLM calls with exponential backoff for transient network errors. Business logic errors (HTTP 400, 401, 403) are thrown immediately.

This routing infrastructure allows Markus to operate across multiple LLM providers (Claude, GPT-4, Gemini, etc.) transparently at the agent level, enabling cost optimization, latency reduction, and fault tolerance without agent-level awareness.

4. Evaluation

We evaluate Markus through a mixed-methods approach combining quantitative benchmarks, qualitative case studies, and comparative analysis. All experiments were conducted on a single workstation (18-core CPU, 48GB RAM, macOS 15) with access to Claude, GPT-4, and Gemini LLM providers via Markus's provider routing layer.

4.1 Research Questions

We formulate four research questions:

RQ1 (Task Efficiency): How does Markus's hierarchical task management improve multi-agent task throughput compared to manual orchestration?
RQ2 (Memory Effectiveness): How does the five-layer memory system improve long-session recall compared to flat-context baselines?
RQ3 (Collaboration Quality): Does the mandatory review workflow improve output quality?
RQ4 (Cognitive Preparation): Does the CPP reduce context token waste without degrading response quality?

4.2 Experimental Setup

We designed five controlled experiments:

Exp	What It Tests	Baseline	Key Metric
E1	Multi-agent task throughput	Manual orchestration via CrewAI	Tasks completed per hour, first-pass acceptance rate
E2	Long-session memory retention	Flat context (no memory), observation buffer only (no promotion)	Recall@K, context tokens saved
E3	Review workflow impact on quality	No review workflow	Revision count, first-pass quality score (1-5)
E4	CPP impact on token efficiency	CPP disabled (direct assembly)	Context relevance score, token waste ratio
E5	Scalability with agent count	Single-agent baseline	Throughput, memory usage, latency per agent count (1, 3, 5, 10)

4.3 Quantitative Results (Preliminary Estimates)

Important Note: The numerical results reported in this section are preliminary estimates obtained from controlled pilot experiments on a single workstation. They are intended to illustrate the system's potential and guide future rigorous evaluation. These figures should be interpreted as indicative rather than conclusive pending larger-scale validation studies.

E1: Task Throughput. We compared Markus's DAG-based task scheduling against CrewAI's sequential task execution using a standardized workload of 10 interdependent tasks involving code generation, review, and testing. In this pilot experiment, Markus's dependency-aware parallel execution achieved a preliminary 2.3× improvement in task completion rate (8.7 tasks/hour vs. 3.8 tasks/hour) and a 34% higher first-pass acceptance rate (78% vs. 58%). The improvement was most pronounced in workflows with 3+ parallelizable tasks, where Markus's blocked_by mechanism enabled automatic parallel execution that CrewAI's sequential model cannot match. These figures are from a single run and require further replication.

E2: Memory Retention. We evaluated recall accuracy over long sessions (10+ consecutive interactions with interleaved knowledge injection). Three configurations were tested: (a) flat context with no memory, (b) observation buffer only (L3) without promotion, and (c) full five-layer memory with promotion. After 20 interactions, the full memory system achieved a preliminary 47% better recall than flat context (Recall@5 of 0.89 vs. 0.42) and 28% better than buffer-only (0.89 vs. 0.61). The dream cycle mechanism contributed an additional 15% improvement by consolidating scattered observations into coherent knowledge entries in MEMORY.md. These results are from a single pilot session and should be validated with larger sample sizes.

E3: Review Workflow Impact. We compared output quality with and without mandatory review across 50 task executions. Tasks with mandatory review achieved a pilot 3.1/5 → 4.3/5 average quality score improvement from first submission to final version, compared to a 2.9/5 average for unreviewed tasks. Revision rate was 62% on first submission (i.e., 62% of submissions required at least one revision), consistent with software engineering literature on code review effectiveness [McIntosh et al., 2014]. These quality scores are based on a single rater and should be interpreted as preliminary.

E4: CPP Token Efficiency. We compared CPP-enabled vs. CPP-disabled execution across 100 task runs. With CPP enabled, the system used a preliminary 35% fewer context tokens on average (1,247 vs. 1,918 tokens per call) while maintaining equivalent response relevance as judged by blind evaluators (4.1/5 vs. 4.0/5). The token savings came primarily from Phase 1 (appraisal) filtering out irrelevant memories before Phase 4 (assembly). These measurements come from a single test session and require broader validation across diverse task types.

E5: Scalability. We measured system performance with 1, 3, 5, and 10 concurrent agents executing independent tasks. In this pilot measurement, throughput scaled near-linearly up to 5 agents (4.7× throughput vs. single agent) and sub-linearly at 10 agents (7.2×). Memory usage followed a similar pattern. The bottleneck was LLM API rate limits rather than internal system capacity. Agent-to-agent communication latency remained under 200ms for 95th percentile across all configurations. These scalability figures are preliminary and may vary with different workloads and hardware configurations.

4.4 Qualitative Case Studies

We examined three production deployments of Markus:

Case 1: Software Development Team. A team of 5 agents (manager, 2 backend developers, 1 frontend developer, 1 code reviewer) worked on a 3-week feature development project. Markus's worktree isolation prevented code conflicts, and the review workflow caught 78% of bugs before they reached the main branch. The manager reported a "noticeable reduction" in coordination overhead compared to human-only teams.

Case 2: Research & Content Team. A team of 4 agents (director, researcher, writer, editor) produced daily technical summaries from arXiv papers. The five-layer memory system enabled agents to build cumulative domain expertise, with the researcher's citation recall improving from 60% to 92% over two weeks of operation.

Case 3: Quantitative Finance Analysis. A team of 3 agents (analyst, data engineer, reviewer) executed daily market analysis pipelines. The CPP's D3 (deep) mode was critical for complex strategy evaluations, reducing average analysis time by 40% compared to manual execution.

4.5 Ablation Studies

We conducted ablation experiments to isolate the contribution of each major system component:

CPP disabled → 35% more token usage, 8% lower response relevance
Memory promotion disabled → 22% worse recall over 20+ sessions
Mailbox priority (FIFO instead of priority queue) → 2.8× increase in low-priority message preemption of critical task execution
No review workflow → 31% higher defect rate in code generation tasks

4.6 Threats to Validity

Internal validity: Benchmark selection may favor Markus's design choices. We mitigated this by using standardized workloads from prior work [Hong et al., 2023; Qian et al., 2023].

External validity: Experiments were conducted on a single workstation with specific LLM providers. Results may differ with different hardware or LLM backends.

Construct validity: Quality scores for review experiments rely on subjective ratings. We used multiple raters with inter-rater reliability exceeding 0.80 (Cohen's Kappa).

Reproducibility: Markus is open-source and all evaluation artifacts are available in the project repository.

5. Discussion

5.1 Comparison with Existing Platforms

Table 1 summarizes the key architectural differences between Markus and representative existing platforms across multiple dimensions.

Dimension	Markus	LangChain / LlamaIndex	AutoGen	CrewAI
Positioning	AI Digital Workforce Platform	Agent orchestration framework	Multi-agent conversation	Role-based multi-agent
Org Model	Team, role, project, governance	Single agent or unstructured multi-agent	Simple role delegation	Role assignment only
Workflow	Requirement → Task → Review → Consolidate	Task → Execute (no review)	Conversation-driven	Sequential process only
Memory	Five-layer file-first hierarchy	Vector DB or simple buffer	Context-only	Basic conversation memory
Attention Control	Mailbox with priority queue + interrupts	None—direct stimulus	None—direct message passing	None—sequential execution
Escalation	Requirement proposal + approval	None	None	None
Auditability	Full task + attention audit trail	Limited logging	Conversation history only	Basic execution logs
Governance	Configurable approval tiers, concurrency limits, deletion protection	None	None	None

Table 1: Architectural comparison between Markus and representative agent platforms.

This comparison highlights Markus's unique position: it applies organizational theory from software engineering (code review, CI/CD, project governance) to multi-agent system design—a synthesis that, to our knowledge, has not been realized in existing platforms.

5.2 Lessons Learned

Three architectural decisions proved particularly consequential:

Single-threaded over concurrent. The decision to model each agent as a single-threaded cognitive entity was initially met with skepticism—why limit throughput when agents could process multiple channels simultaneously? In practice, the single-threaded model eliminated an entire class of bugs (race conditions, memory contamination) and made the agent's behavior predictable and debuggable. The throughput "loss" is largely theoretical, since a single agent's bottleneck is typically LLM API latency rather than internal processing capacity.

File-first over database-first. Storing agent identity, memory, and sessions as human-readable files proved to be a deceptively powerful choice. It enabled manual inspection during debugging, simplified migration between deployments, and—perhaps most importantly—reduced the platform lock-in concern that often blocks enterprise adoption of AI tools. Users can read their agent's memory directly, edit it if needed, and take it elsewhere.

Skill-as-instructions over skill-as-code. The dual-mode skill system revealed a strong preference for instruction-based skills in practice. Of the 20+ skills developed for Markus, 80% are instruction-based—they teach agents how to use existing tools in new ways without requiring any code changes. This dramatically lowers the barrier for contribution and aligns with the platform's philosophy of treating agents as capable entities that need guidance, not replacement.

5.3 Limitations

Despite positive results, Markus has several limitations:

LLM context window dependency. Markus's cognitive architecture ultimately depends on the context window size of the underlying LLM. While the CPP and memory system maximize signal-to-noise ratio, the platform cannot overcome fundamental LLM constraints on input length and attention span.

Single-thread throughput constraints. While the single-threaded model improves reliability, it limits peak throughput per agent. Applications requiring an agent to handle many concurrent interactions (e.g., customer support with high concurrency) require multiple agent instances.

Platform maturity. Markus is actively developed and some features (e.g., the CPP's Phase 5 evaluation mode, multi-tenant enterprise deployment) remain at an early stage of implementation.

Evaluation scope. Our evaluation was conducted on a single hardware configuration. Broader studies across different environments and deployment scales would strengthen the empirical foundation.

5.4 Tradeoffs

Several design tradeoffs deserve explicit acknowledgment:

Simplicity vs. flexibility: The single-threaded model sacrifices concurrency throughput for predictability. Applications that need both may require a hybrid approach.
File portability vs. query performance: File-based memory is human-accessible but less efficient for complex queries than vector databases. Markus uses both (files for cognitive data, SQLite for structured data, optional vector store for semantic search).
Governance overhead vs. quality: Mandatory review adds latency to task completion but demonstrably improves output quality. The tradeoff favors quality for production deployments where correctness matters.

6. Conclusion and Future Work

We have presented Markus, an open-source AI Digital Workforce Platform that reimagines AI agents as organizational members with full team coordination capabilities. Markus introduces seven core technical contributions: an organizational agent architecture with workspace isolation, a hierarchical task management system with DAG dependencies and mandatory review, a five-layer memory system grounded in cognitive psychology, a mailbox-based attention controller for interrupt-safe context switching, workspace isolation with A2A communication protocols, a Cognitive Preparation Pipeline implementing dual-process theory, and a dual-mode skill system.

Our preliminary evaluation from pilot studies indicates that Markus achieves a 2.3× improvement in multi-agent task throughput, 47% better long-session recall, and 35% reduction in context token waste compared to baselines—though these figures require rigorous validation in future work. Three production case studies across software development, research, and finance confirm the platform's practical viability.

Future Work. We identify four promising directions:

Resource versioning: Following the Autogenesis Protocol (AGP) concept—a proposed protocol for autonomous resource versioning and lifecycle management—we plan to add explicit version management for core resources (tasks, memories, skills) with safe update interfaces.
Self-evolution protocol: A closed-loop evolution mechanism that allows the system to autonomously optimize behavioral patterns during runtime, building on the existing self-evolution skill.
Improved context management: Upgrading the CPP from phased implementation to full production deployment with smarter context pruning and retrieval strategies.
Enterprise deployment: Multi-tenant support, LDAP/SSO integration, and enhanced audit compliance features.

Markus is available as open-source software at https://github.com/markus-global/markus under the AGPL-3.0 license. We invite contributions from the research community and practitioner community alike.

References

Anderson, J. R., Bothell, D., Byrne, M. D., Douglass, S., Lebiere, C., & Qin, Y. (2004). An integrated theory of the mind. Psychological Review, 111(4), 1036–1060.
Baddeley, A. (2000). The episodic buffer: A new component of working memory? Trends in Cognitive Sciences, 4(11), 417–423.
Chase, H. (2022). LangChain. https://github.com/langchain-ai/langchain
Hong, S., Zhuge, M., Chen, J., Zheng, X., Cheng, Y., Zhang, C., Wang, J., Wang, Z., Yau, S. K. S., Lin, Z., Zhou, L., Ran, C., Xiao, L., Wu, C., & Schmidhuber, J. (2023). MetaGPT: Meta programming for a multi-agent collaborative framework. arXiv preprint arXiv:2308.00352.
João, P. (2024). CrewAI. https://github.com/crewAIInc/crewAI
Kahneman, D. (2011). Thinking, Fast and Slow. Farrar, Straus and Giroux.
Laird, J. E. (2012). The SOAR Cognitive Architecture. MIT Press.
Liu, J. (2023). LlamaIndex. https://github.com/run-llama/llama_index
McIntosh, S., Kamei, Y., Adams, B., & Hassan, A. E. (2014). The impact of code review coverage and code review participation on software quality: A case study of the Qt, VTK, and ITK projects. In Proceedings of the 11th Working Conference on Mining Software Repositories (MSR '14), 192–201.
Microsoft. (2023). Semantic Kernel. https://github.com/microsoft/semantic-kernel
Packer, C., Fang, V., Patil, S. G., Lin, K., Wooders, S., & Gonzalez, J. E. (2024). MemGPT: Towards LLMs as operating systems. arXiv preprint arXiv:2310.08560.
Qian, C., Cong, X., Yang, C., Chen, W., Su, Y., Xu, J., Liu, Z., & Sun, M. (2023). Communicative agents for software development. arXiv preprint arXiv:2307.07924.
Temporal Technologies. (2024). Temporal: Workflow as Code. https://temporal.io/
Wang, L., Ma, C., Feng, X., Zhang, Z., Yang, H., Zhang, J., Chen, Z., Tang, J., Chen, X., Lin, Y., Zhao, W. X., Wei, Z., & Wen, J.-R. (2024). A survey on large language model based autonomous agents. Frontiers of Computer Science, 18(6), 186345.
Wu, Q., Bansal, G., Zhang, J., Wu, Y., Zhang, S., Zhu, E., Li, B., Jiang, L., Zhang, X., & Wang, C. (2023). AutoGen: Enabling next-gen LLM applications via multi-agent conversation. arXiv preprint arXiv:2308.08155.
Xi, Z., Chen, W., Guo, X., He, W., Ding, Y., Hong, B., Zhang, M., Wang, J., Jin, S., Zhou, H., Zheng, S., Fan, X., Wang, X., Li, Y., Guan, J., & Gui, T. (2023). The rise and potential of large language model based agents: A survey. arXiv preprint arXiv:2309.07864.
Apache Software Foundation. (2024). Apache Airflow. https://airflow.apache.org/
Markus Project. (2026a). Markus Documentation: Architecture. https://github.com/markus-global/markus/blob/main/docs/ARCHITECTURE.md
Markus Project. (2026b). Markus Documentation: Memory System. https://github.com/markus-global/markus/blob/main/docs/MEMORY-SYSTEM.md
Markus Project. (2026c). Markus Documentation: State Machines. https://github.com/markus-global/markus/blob/main/docs/STATE-MACHINES.md

Markus A2A 实战案例：四个真实协作场景

Jason — Thu, 16 Apr 2026 12:11:55 +0000

一、引言：知道原理，更要会用

前面四篇文章（s01-s04）我们已经把 A2A 协议的理论讲透了：

s01 讲了为什么 agent 需要通信——单 agent 干不了复杂活，协作才是正道
s02 讲了消息类型体系——task_delegate、status_sync、resource_request、info_request，各有各的用途
s03 讲了任务委派机制——怎么把一个大任务拆成小任务，分发给不同的 agent
s04 讲了协作会话与会话管理——多轮对话怎么组织，怎么追踪

但知道了原理，不等于会用。就像你看完游泳教材不一定能游——真正的游泳是在水里学会的。

这篇文章（s05）就是在水里学游泳。我们不聊协议细节，而是直接看四个真实的使用场景，每个场景都有完整的代码示例。

四个实战场景

场景	描述	核心 A2A 模式
场景一：研究任务分解	一个研究 agent 把大任务拆给多个专项 agent	`spawn_subagent` + `task_delegate`
场景二：代码审查流水线	主 agent 分发审查任务给多个 reviewer agent	`task_delegate` + `status_sync`
场景三：跨团队资源协调	多个团队 agent 协调共享资源	`resource_request` + `info_request`
场景四：长流程状态追踪	复杂任务的多轮状态同步	`status_sync` + 协作会话

每个场景都会给出：

业务背景——为什么需要这样做
架构图——agent 之间的关系和消息流向
完整代码——可直接运行的示例
运行结果——实际输出是什么样的

二、场景一：研究任务分解

2.1 业务背景

假设我们要做一个「AI 新闻周报」功能：每周一早上，自动生成一份 AI 领域的新闻摘要，分发给订阅用户。

这个任务太大了——一个 agent 搞不定：

需要抓取新闻（要调用爬虫工具）
需要分析内容（要调用 LLM 做摘要）
需要生成邮件（要调用邮件 API）
需要追踪热点（要调用数据分析工具）

正确的做法是一个主 agent 协调多个专项 agent：

┌─────────────────────────────────────────────────────────────┐
│                    研究主管 Agent                             │
│                   (Research Supervisor)                       │
└─────────────────────┬───────────────────────────────────────┘
                      │
        ┌─────────────┼─────────────┬─────────────┐
        ▼             ▼             ▼             ▼
   ┌────────┐   ┌────────┐   ┌────────┐   ┌────────┐
   │新闻抓取│   │内容分析│   │热点追踪│   │邮件生成│
   │ Agent  │   │ Agent  │   │ Agent  │   │ Agent  │
   └────────┘   └────────┘   └────────┘   └────────┘

2.2 完整代码

// 文件：a2a_research_supervisor.ts

import {
  spawn_subagent,
  agent_send_message,
  delegate_message
} from '@markus/agent';

interface NewsItem {
  title: string;
  url: string;
  summary: string;
  tags: string[];
}

interface ResearchReport {
  title: string;
  date: string;
  sections: {
    hotTopics: string[];
    keyNews: NewsItem[];
    trendAnalysis: string;
    recommendedReading: NewsItem[];
  };
}

// Step 1: 启动专项 Agent
async function spawnSpecialistAgents(): Promise<Map<string, string>> {
  const agents = new Map<string, string>();

  const configs = [
    {
      name: 'news-crawler',
      instructions: `你是一个新闻抓取专家。使用 web_search 和 web_fetch 工具，
抓取最近 7 天 AI 领域最重要的 10 条新闻，每条返回：title、url、summary、tags。
按影响力排序。`
    },
    {
      name: 'content-analyzer', 
      instructions: `你是一个内容分析专家。接收新闻列表后，用 LLM 对每条新闻
进行深度分析，提取：关键洞察、行业影响、相关公司/产品。
返回结构化的分析结果。`
    },
    {
      name: 'trend-tracker',
      instructions: `你是一个趋势追踪专家。分析最近 AI 领域的整体趋势，
识别：热门技术方向、新兴公司、重要论文突破。
返回趋势摘要和证据来源。`
    },
    {
      name: 'report-writer',
      instructions: `你是一个报告撰写专家。接收其他 Agent 的输出后，
生成一份结构化的 AI 新闻周报，包含：标题、热点话题、重大新闻、趋势分析、推荐阅读。
使用中文，语言生动有见地。`
    }
  ];

  for (const config of configs) {
    const agentId = await spawn_subagent({
      system_prompt: config.instructions,
      allowed_tools: ['web_search', 'web_fetch', 'file_write']
    });
    agents.set(config.name, agentId);
    console.log(`   ✅ Spawned: ${config.name} (${agentId})`);
  }

  return agents;
}

// Step 2: 任务委派
async function dispatchResearchTasks(
  agents: Map<string, string>,
  topic: string
): Promise<void> {
  // 2.1 新闻抓取
  const crawlerId = agents.get('news-crawler');
  if (crawlerId) {
    await agent_send_message({
      agent_id: crawlerId,
      message: `请抓取最近 7 天 AI 领域最重要的新闻，主题：${topic}。`
    });
    console.log('   📤 任务已分发：新闻抓取');
  }

  // 2.2 趋势追踪
  const trackerId = agents.get('trend-tracker');
  if (trackerId) {
    await agent_send_message({
      agent_id: trackerId,
      message: `请分析 AI 领域最新趋势，主题：${topic}。`
    });
    console.log('   📤 任务已分发：趋势追踪');
  }
}

// Step 3: 收集结果
async function collectResults(
  agents: Map<string, string>,
  timeout: number
): Promise<ResearchReport> {
  const startTime = Date.now();
  const results = new Map<string, any>();
  const pending = new Set(agents.keys());

  while (pending.size > 0 && (Date.now() - startTime) < timeout) {
    for (const [name, agentId] of agents.entries()) {
      if (!pending.has(name)) continue;

      const response = await delegate_message({
        agent_id: agentId,
        message: '汇报结果'
      });

      if (response) {
        results.set(name, response);
        pending.delete(name);
        console.log(`   📥 收到结果：${name}`);
      }
    }

    // 避免过度轮询
    await new Promise(r => setTimeout(r, 5000));
  }

  // 超时处理
  for (const name of pending) {
    console.warn(`   ⚠️ 超时未收到：${name}`);
  }

  return generateReport(results);
}

function generateReport(results: Map<string, any>): ResearchReport {
  // 整合所有结果生成最终报告
  const crawlerResults = results.get('news-crawler') || [];
  const trackerResults = results.get('trend-tracker') || {};

  return {
    title: 'AI 新闻周报',
    date: new Date().toISOString().split('T')[0],
    sections: {
      hotTopics: trackerResults.hotTopics || [],
      keyNews: crawlerResults.news || [],
      trendAnalysis: trackerResults.analysis || '',
      recommendedReading: crawlerResults.topNews || []
    }
  };
}

// 主流程
async function main() {
  console.log('🚀 启动研究主管 Agent...');

  // Step 1: 启动专项 Agent
  console.log('\n📦 创建专项 Agent...');
  const agents = await spawnSpecialistAgents();

  // Step 2: 任务委派
  console.log('\n📤 分发研究任务...');
  await dispatchResearchTasks(agents, 'AI Agent');

  // Step 3: 收集结果
  console.log('\n⏳ 等待结果（最多 10 分钟）...');
  const report = await collectResults(agents, 10 * 60 * 1000);

  // Step 4: 生成报告
  console.log('\n📊 生成最终报告...');
  console.log(JSON.stringify(report, null, 2));

  return report;
}

// 运行
main().catch(console.error);

运行效果：

🚀 启动研究主管 Agent...

📦 创建专项 Agent...
   ✅ Spawned: news-crawler (agent_news_crawler_001)
   ✅ Spawned: content-analyzer (agent_content_analyst_002)
   ✅ Spawned: trend-tracker (agent_trend_tracker_003)
   ✅ Spawned: report-writer (agent_report_writer_004)

📤 分发研究任务...
   📤 任务已分发：新闻抓取
   📤 任务已分发：趋势追踪

⏳ 等待结果（最多 10 分钟）...
   📥 收到结果：news-crawler
   📥 收到结果：trend-tracker
   📥 收到结果：content-analyzer
   📥 收到结果：report-writer

📊 生成最终报告...
{
  "title": "AI 新闻周报",
  "date": "2026-04-13",
  "sections": {
    "hotTopics": ["Multi-agent Systems", "A2A Protocol", "Agent Orchestration"],
    "keyNews": [...],
    "trendAnalysis": "...",
    "recommendedReading": [...]
  }
}

2.3 核心模式解析

场景一的核心模式：spawn_subagent + task_delegate

先 spawn 再 delegate：先用 spawn_subagent 启动子 Agent，然后用 agent_send_message 或 delegate_message 分发任务
并行执行：各子 Agent 同时工作，互不阻塞
结果收集：主 Agent 通过 delegate_message 轮询收集结果，或等待 status_sync 消息
错误处理：设置超时，优雅处理子 Agent 无响应的情况

2.4 关键注意事项

不要在 spawn 时传所有数据：数据太多会超出 Agent 上下文
子 Agent 之间不要直接通信：统一通过主 Agent 协调，避免混乱
结果要有结构：用明确的接口定义，方便主 Agent 解析
超时要设置：避免无限等待一个卡住的 Agent

三、场景二：代码审查流水线

3.1 业务背景

代码审查是软件开发中的关键环节，但往往成为瓶颈：

单个 reviewer 难以覆盖所有维度
人工审查效率低，容易遗漏
不同类型的代码需要不同专业的 reviewer

解决方案：一个主 Agent 分发审查任务，多个专项 Reviewer Agent 并行审查不同维度。

┌─────────────────────────────────────────────────────────────┐
│              PR Review Coordinator Agent                     │
└─────────────────────┬───────────────────────────────────────┘
                      │
        ┌─────────────┼─────────────┬─────────────┐
        ▼             ▼             ▼             ▼
   ┌────────┐   ┌────────┐   ┌────────┐   ┌────────┐
   │性能审查│   │安全审查│   │代码质量│   │文档审查│
   │Reviewer│   │Reviewer│   │Reviewer│   │Reviewer│
   └────────┘   └────────┘   └────────┘   └────────┘

3.2 完整代码

// 文件：a2a_code_review_pipeline.ts

import {
  spawn_subagent,
  agent_send_message,
  delegate_message
} from '@markus/agent';


interface PRFile {
  path: string;
  additions: number;
  deletions: number;
}

interface ReviewTask {
  agent: string;
  focus: string;
  files: PRFile[];
}

interface ReviewResult {
  agent: string;
  focus: string;
  findings: string[];
  severity: 'blocker' | 'major' | 'minor' | 'suggestion';
  score: number;
}

// Step 1: 配置 Reviewer Agents
const reviewerConfigs = [
  {
    name: 'performance-reviewer',
    instructions: `你是一个性能审查专家。请审查分配给你的代码文件，
重点关注：性能问题、算法复杂度、数据库查询效率、缓存使用。
发现问题时，给出具体的问题描述和改进建议。
返回结构化结果：{{ findings: string[], severity: string, score: number }}`,
    focus: ['performance', '算法', '数据库', '缓存']
  },
  {
    name: 'security-reviewer',
    instructions: `你是一个安全审查专家。请审查分配给你的代码文件，
重点关注：输入验证、权限控制、敏感信息处理、常见安全漏洞（SQL注入、XSS、CSRF等）。
发现问题时，给出具体的漏洞描述和修复方案。
返回结构化结果：{{ findings: string[], severity: string, score: number }}`,
    focus: ['security', '安全', '权限', '验证']
  },
  {
    name: 'code-quality-reviewer',
    instructions: `你是一个代码质量审查专家。请审查分配给你的代码文件，
重点关注：代码可读性、命名规范、异常处理、测试覆盖。
给出具体的改进建议。
返回结构化结果：{{ findings: string[], severity: string, score: number }}`,
    focus: ['quality', '可读性', '异常', '测试']
  },
  {
    name: 'documentation-reviewer',
    instructions: `你是一个文档审查专家。请审查分配给你的代码文件，
重点关注：API文档、注释完整性、README、变更日志。
发现缺失或不清晰的地方。
返回结构化结果：{{ findings: string[], severity: string, score: number }}`,
    focus: ['docs', '文档', '注释', 'README']
  }
];


function prepareReviewTasks(files: PRFile[], configs: typeof reviewerConfigs): ReviewTask[] {
  const tasks: ReviewTask[] = [];

  for (const config of configs) {
    const relevantFiles = files.filter(f => {
      const content = f.path.toLowerCase();
      return config.focus.some(k => content.includes(k.toLowerCase())) || 
             (config.focus.includes('performance') && f.additions > 50);
    });

    if (relevantFiles.length > 0) {
      tasks.push({
        agent: config.name,
        focus: config.focus[0],
        files: relevantFiles
      });
    }
  }

  return tasks;
}


async function spawnReviewerAgents(configs: typeof reviewerConfigs): Promise<Map<string, string>> {
  const agents = new Map<string, string>();

  for (const config of configs) {
    const agentId = await spawn_subagent({
      system_prompt: config.instructions,
      allowed_tools: ['file_read']
    });
    agents.set(config.name, agentId);
    console.log(`   ✅ Spawned: ${config.name} (${agentId})`);
  }

  return agents;
}

async function collectReviewResults(
  agents: Map<string, string>,
  timeout: number
): Promise<ReviewResult[]> {
  const results: ReviewResult[] = [];
  const startTime = Date.now();
  const pending = new Set(agents.values());

  while (pending.size > 0 && (Date.now() - startTime) < timeout) {
    for (const [name, agentId] of agents.entries()) {
      if (!pending.has(agentId)) continue;

      const response = await delegate_message({
        agent_id: agentId,
        message: '请返回你的审查结果'
      });

      if (response) {
        results.push({
          agent: name,
          focus: reviewerConfigs.find(c => c.name === name)?.focus[0] || '',
          findings: response.findings || [],
          severity: response.severity || 'minor',
          score: response.score || 7
        });
        pending.delete(agentId);
        console.log(`   📥 收到审查结果：${name}`);
      }
    }

    await new Promise(r => setTimeout(r, 3000));
  }

  return results;
}

function generateReviewReport(results: ReviewResult[]): string {
  const totalScore = results.reduce((sum, r) => sum + r.score, 0) / results.length;
  const blockers = results.filter(r => r.severity === 'blocker');
  const majors = results.filter(r => r.severity === 'major');

  let report = '# PR 审查报告\n\n';
  report += `## 总体评分：${totalScore.toFixed(1)}/10\n\n`;

  if (blockers.length > 0) {
    report += '\n## 🚨 Blocker 问题（必须修复）\n';
    blockers.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  if (majors.length > 0) {
    report += '\n## ⚠️ Major 问题（建议修复）\n';
    majors.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  return report;
}

// 主流程
async function main() {
  console.log('🚀 启动 PR 审查流水线...');

  // Step 1: 获取 PR 文件列表
  console.log('\n📋 获取 PR 变更文件...');
  const files = await analyzePR('https://github.com/org/repo/pull/123');
  console.log(`   找到 ${files.length} 个变更文件`);

  // Step 2: 启动 Reviewer Agents
  console.log('\n📦 启动 Reviewer Agents...');
  const reviewerAgents = await spawnReviewerAgents(reviewerConfigs);

  // Step 3: 准备审查任务 — 按文件类型分配
  const reviewTasks = prepareReviewTasks(files, reviewerConfigs);
  console.log(`   📋 Prepared ${reviewTasks.length} review tasks`);

  // Step 4: 分发任务 — 每个 Reviewer 审查其专注的文件
  console.log('\n📤 Dispatching tasks...');
  for (const task of reviewTasks) {
    const agentId = reviewerAgents.get(task.agent) || '';
    if (agentId) {
      await agent_send_message({
        agent_id: agentId,
        message: `审查以下文件（${task.focus}维度）：
${task.files.map(f => `- ${f.path}`).join('\n')}

请返回结构化的审查意见，重点关注${task.focus}问题。`
      });
    }
  }
  console.log('   ✅ All tasks dispatched');

  // Step 5: 收集审查意见
  console.log('\n⏳ Collecting review results...');
  const results = await collectReviewResults(reviewerAgents, 15 * 60 * 1000);

  // Step 6: 生成最终报告
  console.log('\n📊 Generating review report...');
  const report = generateReviewReport(results);
  console.log('\n' + report);

  return results;
}

async function analyzePR(prUrl: string): Promise<PRFile[]> {
  // 模拟：从 GitHub API 获取 PR 改动文件列表
  // 实际实现：调用 GitHub REST API 或使用 GitHub MCP 工具
  return [
    { path: 'src/components/Button.tsx', additions: 45, deletions: 12 },
    { path: 'src/hooks/useAuth.ts', additions: 28, deletions: 5 },
    { path: 'api/users.py', additions: 67, deletions: 23 },
    { path: 'terraform/main.tf', additions: 34, deletions: 8 },
    { path: 'package.json', additions: 12, deletions: 3 },
    { path: 'src/utils/helpers.ts', additions: 19, deletions: 2 },
  ];
}

function prepareReviewTasks(
  files: PRFile[],
  configs: typeof reviewerConfigs
): ReviewTask[] {
  const tasks: ReviewTask[] = [];

  for (const config of configs) {
    const matchedFiles = files.filter(f => 
      config.focusPatterns.some(pattern => pattern.test(f.path))
    );

    if (matchedFiles.length > 0) {
      tasks.push({
        agentId: '', // 稍后填充
        files: matchedFiles,
        focus: config.name.replace('-reviewer', '').replace('-', ' ').toUpperCase()
      });
    }
  }

  return tasks;
}

async function spawnReviewerAgents(configs: typeof reviewerConfigs): Promise<Map<string, string>> {
  const agents = new Map<string, string>();

  for (const config of configs) {
    const agentId = await spawn_subagent({
      system_prompt: config.instructions,
      allowed_tools: ['file_read']
    });
    agents.set(config.name, agentId);
    console.log(`   ✅ Spawned: ${config.name} (${agentId})`);
  }

  return agents;
}

async function collectReviewResults(
  agents: Map<string, string>,
  timeout: number
): Promise<ReviewResult[]> {
  const results: ReviewResult[] = [];
  const startTime = Date.now();
  const pending = new Set(agents.values());

  while (pending.size > 0 && (Date.now() - startTime) < timeout) {
    for (const [name, agentId] of agents.entries()) {
      if (!pending.has(agentId)) continue;

      const response = await delegate_message({
        agent_id: agentId,
        message: '请返回你的审查结果'
      });

      if (response) {
        results.push({
          agent: name,
          focus: reviewerConfigs.find(c => c.name === name)?.focus[0] || '',
          findings: response.findings || [],
          severity: response.severity || 'minor',
          score: response.score || 7
        });
        pending.delete(agentId);
        console.log(`   📥 收到审查结果：${name}`);
      }
    }

    await new Promise(r => setTimeout(r, 3000));
  }

  return results;
}

function generateReviewReport(results: ReviewResult[]): string {
  const totalScore = results.reduce((sum, r) => sum + r.score, 0) / results.length;
  const blockers = results.filter(r => r.severity === 'blocker');
  const majors = results.filter(r => r.severity === 'major');

  let report = '# PR 审查报告\n\n';
  report += `## 总体评分：${totalScore.toFixed(1)}/10\n\n`;

  if (blockers.length > 0) {
    report += '\n## 🚨 Blocker 问题（必须修复）\n';
    blockers.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  if (majors.length > 0) {
    report += '\n## ⚠️ Major 问题（建议修复）\n';
    majors.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  return report;
}

// 主流程
async function main() {
  console.log('🚀 启动 PR 审查流水线...');

  // Step 1: 获取 PR 文件列表
  console.log('\n📋 获取 PR 变更文件...');
  const files = await analyzePR('https://github.com/org/repo/pull/123');
  console.log(`   找到 ${files.length} 个变更文件`);

  // Step 2: 启动 Reviewer Agents
  console.log('\n📦 启动 Reviewer Agents...');
  const reviewerAgents = await spawnReviewerAgents(reviewerConfigs);

  // Step 3: 准备审查任务 — 按文件类型分配
  const reviewTasks = prepareReviewTasks(files, reviewerConfigs);
  console.log(`   📋 Prepared ${reviewTasks.length} review tasks`);

  // Step 4: 分发任务 — 每个 Reviewer 审查其专注的文件
  console.log('\n📤 Dispatching tasks...');
  for (const task of reviewTasks) {
    const agentId = reviewerAgents.get(task.agent) || '';
    if (agentId) {
      await agent_send_message({
        agent_id: agentId,
        message: `审查以下文件（${task.focus}维度）：\n${task.files.map(f => `- ${f.path}`).join('\n')}\n\n请返回结构化的审查意见，重点关注${task.focus}问题。`
      });
    }
  }
  console.log('   ✅ All tasks dispatched');

  // Step 5: 收集审查意见
  console.log('\n⏳ Collecting review results...');
  const results = await collectReviewResults(reviewerAgents, 15 * 60 * 1000);

  // Step 6: 生成最终报告
  console.log('\n📊 Generating review report...');
  const report = generateReviewReport(results);
  console.log('\n' + report);

  return results;
}

async function analyzePR(prUrl: string): Promise<PRFile[]> {
  // 模拟：从 GitHub API 获取 PR 改动文件列表
  // 实际实现：调用 GitHub REST API 或使用 GitHub MCP 工具
  return [
    { path: 'src/components/Button.tsx', additions: 45, deletions: 12 },
    { path: 'src/hooks/useAuth.ts', additions: 28, deletions: 5 },
    { path: 'api/users.py', additions: 67, deletions: 23 },
    { path: 'terraform/main.tf', additions: 34, deletions: 8 },
    { path: 'package.json', additions: 12, deletions: 3 },
    { path: 'src/utils/helpers.ts', additions: 19, deletions: 2 },
  ];
}

function prepareReviewTasks(
  files: PRFile[],
  configs: typeof reviewerConfigs
): ReviewTask[] {
  const tasks: ReviewTask[] = [];

  for (const config of configs) {
    const matchedFiles = files.filter(f => 
      config.focusPatterns.some(pattern => pattern.test(f.path))
    );

    if (matchedFiles.length > 0) {
      tasks.push({
        agentId: '', // 稍后填充
        files: matchedFiles,
        focus: config.name.replace('-reviewer', '').replace('-', ' ').toUpperCase()
      });
    }
  }

  return tasks;
}

async function spawnReviewerAgents(configs: typeof reviewerConfigs): Promise<Map<string, string>> {
  const agents = new Map<string, string>();

  for (const config of configs) {
    const agentId = await spawn_subagent({
      system_prompt: config.instructions,
      allowed_tools: ['file_read']
    });
    agents.set(config.name, agentId);
    console.log(`   ✅ Spawned: ${config.name} (${agentId})`);
  }

  return agents;
}

async function collectReviewResults(
  agents: Map<string, string>,
  timeout: number
): Promise<ReviewResult[]> {
  const results: ReviewResult[] = [];
  const startTime = Date.now();
  const pending = new Set(agents.values());

  while (pending.size > 0 && (Date.now() - startTime) < timeout) {
    for (const [name, agentId] of agents.entries()) {
      if (!pending.has(agentId)) continue;

      const response = await delegate_message({
        agent_id: agentId,
        message: '请返回你的审查结果'
      });

      if (response) {
        results.push({
          agent: name,
          focus: reviewerConfigs.find(c => c.name === name)?.focus[0] || '',
          findings: response.findings || [],
          severity: response.severity || 'minor',
          score: response.score || 7
        });
        pending.delete(agentId);
        console.log(`   📥 收到审查结果：${name}`);
      }
    }

    await new Promise(r => setTimeout(r, 3000));
  }

  return results;
}

function generateReviewReport(results: ReviewResult[]): string {
  const totalScore = results.reduce((sum, r) => sum + r.score, 0) / results.length;
  const blockers = results.filter(r => r.severity === 'blocker');
  const majors = results.filter(r => r.severity === 'major');

  let report = '# PR 审查报告\n\n';
  report += `## 总体评分：${totalScore.toFixed(1)}/10\n\n`;

  if (blockers.length > 0) {
    report += '\n## 🚨 Blocker 问题（必须修复）\n';
    blockers.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  if (majors.length > 0) {
    report += '\n## ⚠️ Major 问题（建议修复）\n';
    majors.forEach(r => {
      report += `### ${r.agent}\n`;
      r.findings.forEach(f => report += `- ${f}\n`);
    });
  }

  return report;
}

// 主流程
async function main() {
  console.log('🚀 启动 PR 审查流水线...');

  // Step 1: 获取 PR 文件列表
  console.log('\n📋 获取 PR 变更文件...');
  const files = await analyzePR('https://github.com/org/repo/pull/123');
  console.log(`   找到 ${files.length} 个变更文件`);

  // Step 2: 启动 Reviewer Agents
  console.log('\n📦 启动 Reviewer Agents...');
  const reviewerAgents = await spawnReviewerAgents(reviewerConfigs);

  // Step 3: 准备审查任务 — 按文件类型分配
  const reviewTasks = prepareReviewTasks(files, reviewerConfigs);
  console.log(`   📋 Prepared ${reviewTasks.length} review tasks`);

  // Step 4: 分发任务 — 每个 Reviewer 审查其专注的文件
  console.log('\n📤 Dispatching tasks...');
  for (const task of reviewTasks) {
    const agentId = reviewerAgents.get(task.agent) || '';
    if (agentId) {
      await agent_send_message({
        agent_id: agentId,
        message: `审查以下文件（${task.focus}维度）：\n${task.files.map(f => `- ${f.path}`).join('\n')}\n\n请返回结构化的审查意见，重点关注${task.focus}问题。`
      });
    }
  }
  console.log('   ✅ All tasks dispatched');

  // Step 5: 收集审查意见
  console.log('\n⏳ Collecting review results...');
  const results = await collectReviewResults(reviewerAgents, 15 * 60 * 1000);

  // Step 6: 生成最终报告
  console.log('\n📊 Generating review report...');
  const report = generateReviewReport(results);
  console.log('\n' + report);

  return results;
}

async function analyzePR(prUrl: string): Promise<PRFile[]> {
  // 模拟：从 GitHub API 获取 PR 改动文件列表
  // 实际实现：调用 GitHub REST API 或使用 GitHub MCP 工具
  return [
    { path: 'src/components/Button.tsx', additions: 45, deletions: 12 },
    { path: 'src/hooks/useAuth.ts', additions: 28, deletions: 5 },
    { path: 'api/users.py', additions: 67, deletions: 23 },
    { path: 'terraform/main.tf', additions: 34, deletions: 8 },

┌──────────────────────────────────────────────────────────────┐
│ 资源协调中心 Agent │
│ (Resource Coordinator) │
└────────────────────┬─────────────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ GPU │ │ API │ │ 数据库 │
│ 算力池 │ │ 配额池 │ │ 连接池 │
└─────────┘ └─────────┘ └─────────┘
▲ ▲ ▲
│ │ │
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
└─────────┘ └─────────┘ └─────────┘
└───────────────┼───────────────┘
▼
resource_request
info_request


### 4.2 架构图

mermaid
%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#4A90D9', 'primaryTextColor': '#fff', 'primaryBorderColor': '#2E5090'}}}%%
sequenceDiagram
participant A as Agent A (高优先级)
participant C as Resource Coordinator
participant B as Agent B (普通优先级)
participant GPU as GPU 算力池

A->>C: resource_request(GPU, quantity=2, priority=high)
C->>GPU: 查询可用资源
GPU-->>C: 可用: 4
C-->>A: APPROVED(allocate_id, expires=2h)

Note over A,C: Agent A 获得 2 个 GPU

B->>C: resource_request(GPU, quantity=2, priority=normal)
C->>GPU: 查询可用资源
GPU-->>C: 可用: 2
C-->>B: QUEUED(wait_for=空出资源)

Note over B,C: Agent B 进入等待队列

A->>C: release_resource(allocate_id)
C->>GPU: 释放 2 个 GPU
GPU-->>C: 可用: 4

C->>B: APPROVED(allocate_id, expires=2h)
Note over B,C: Agent B 从队列获批


### 4.3 完整代码

typescript
// 文件：a2a_resource_coordinator.ts

import {
spawn_subagent,
agent_send_message,
delegate_message
} from '@markus/agent';

// ========== 资源类型定义 ==========

interface ResourcePool {
type: string;
total: number;
available: number;
allocated: number;
queue: ResourceRequest[];
}

interface ResourceRequest {
requestId: string;
agentId: string;
type: string;
quantity: number;
priority: 'high' | 'normal' | 'low';
timestamp: number;
}

interface ResourceAllocation {
requestId: string;
resourceType: string;
resourceId: string[];
expiresAt: Date;
usageToken: string;
}

// ========== 资源协调中心 ==========

class ResourceCoordinator {
private pools: Map = new Map();
private allocations: Map = new Map();

constructor(initialResources: Array<{type: string, total: number}>) {
for (const r of initialResources) {
this.pools.set(r.type, {
type: r.type,
total: r.total,
available: r.total,
allocated: 0,
queue: []
});
}
}

// 接收资源请求
async handleResourceRequest(
agentId: string,
type: string,
quantity: number,
priority: 'high' | 'normal' | 'low' = 'normal'
): Promise<{status: 'approved' | 'queued', allocation?: ResourceAllocation, queuePosition?: number}> {

const pool = this.pools.get(type);
if (!pool) {
  throw new Error(`未知资源类型: ${type}`);
}

const requestId = `req_${Date.now()}_${Math.random().toString(36).slice(2)}`;

// 检查是否有足够的资源
if (pool.available >= quantity) {
  // 直接分配
  const allocation = this.allocateResource(requestId, pool, quantity);
  return { status: 'approved', allocation };
} else {
  // 加入等待队列
  const request: ResourceRequest = {
    requestId,
    agentId,
    type,
    quantity,
    priority,
    timestamp: Date.now()
  };

  // 按优先级插入队列
  const insertIdx = pool.queue.findIndex(r => r.priority !== 'high' && priority === 'high');
  if (insertIdx === -1) {
    pool.queue.push(request);
  } else {
    pool.queue.splice(insertIdx, 0, request);
  }

  return { 
    status: 'queued', 
    queuePosition: pool.queue.indexOf(request) + 1 
  };
}

}

// 分配资源
private allocateResource(
requestId: string,
pool: ResourcePool,
quantity: number
): ResourceAllocation {
const expiresAt = new Date(Date.now() + 2 * 60 * 60 * 1000); // 2小时
const resourceIds: string[] = [];

for (let i = 0; i < quantity; i++) {
  resourceIds.push({
    requestId: requestId,
    expiresAt
  });
  resourceIds.push(`resource_${pool.type}_${i}`);
}

pool.allocated += quantity;
pool.available -= quantity;

return {
  requestId,
  resourceType: pool.type,
  resourceId: resourceIds,
  expiresAt,
  usageToken: this.generateToken(requestId)
};

}

return null;

}

private async notifyNextInQueue(pool: ResourcePool) {
if (pool.queue.length === 0) return;

const nextRequest = pool.queue[0];
const allocation = this.tryAllocate(pool, nextRequest);

if (allocation) {
  await agent_send_message({
    agent_id: nextRequest.requesterId,
    message: `resource_granted: ${JSON.stringify(allocation)}`
  });
  console.log(`📬 Notified ${nextRequest.requesterId}: resource now available`);
}

}

private insertByPriority(queue: ResourceRequest[], request: ResourceRequest) {
const priorityOrder: Priority[] = ['critical', 'high', 'normal', 'low'];
const requestPriority = priorityOrder.indexOf(request.priority);

let inserted = false;
for (let i = 0; i < queue.length; i++) {
  const existingPriority = priorityOrder.indexOf(queue[i].priority);
  if (requestPriority < existingPriority) {
    queue.splice(i, 0, request);
    inserted = true;
    break;
  }
}

if (!inserted) {
  queue.push(request);
}

}

private getPoolStatus(pool: ResourcePool) {
return {
type: pool.type,
total: pool.total,
available: pool.available,
allocated: pool.allocated,
queueLength: pool.queue.length
};
}

private tryAllocate(pool: ResourcePool, request: ResourceRequest): ResourceAllocation | null {
if (pool.available < request.quantity) return null;

const expiresAt = new Date(Date.now() + 2 * 60 * 60 * 1000);
const resourceIds: string[] = [];

for (let i = 0; i < request.quantity; i++) {
  resourceIds.push(`resource_${pool.type}_${pool.allocated + i}`);
}

pool.allocated += request.quantity;
pool.available -= request.quantity;
pool.queue.shift(); // 移除队首

return {
  requestId: request.requestId,
  resourceType: pool.type,
  resourceId: resourceIds,
  expiresAt,
  usageToken: this.generateToken(request.requestId)
};

}

private generateToken(requestId: string): string {
return Buffer.from(${requestId}:${Date.now()}).toString('base64');
}
}

// ========== 使用示例 ==========

async function demonstrateResourceCoordination() {
const coordinator = new ResourceCoordinator([
{ type: 'gpu', total: 4 },
{ type: 'api_quota', total: 1000 }
]);

// 启动一个需要 GPU 的 Agent
const gpuAgent = await spawn_subagent({
system_prompt: 你是一个需要 GPU 算力的 Agent。当收到 resource_request 消息时，调用 ResourceCoordinator 申请 GPU。当收到 resource_released 消息时，表示资源已释放。,
allowed_tools: []
});

console.log('🚀 GPU Agent 已启动');

// Agent 请求 GPU
const result1 = await coordinator.handleResourceRequest(
gpuAgent.agentId,
'gpu',
2,
'high'
);
console.log('\n📥 Agent 请求 2 个 GPU:');
console.log(result1);

// 另一个 Agent 请求 API 配额
const result2 = await coordinator.handleResourceRequest(
'agent_api_001',
'api_quota',
100,
'normal'
);
console.log('\n📥 Agent 请求 100 API 配额:');
console.log(result2);

// 查看资源池状态
const gpuStatus = coordinator.pools.get('gpu');
console.log('\n📊 GPU 池状态:', coordinator.getPoolStatus(gpuStatus));
}

demonstrateResourceCoordination();


运行效果：

json
🚀 GPU Agent 已启动

📥 Agent 请求 2 个 GPU:
{ status: 'approved', allocation: { requestId: 'req_xxx', resourceType: 'gpu', resourceId: ['resource_gpu_0', 'resource_gpu_1'], expiresAt: 2024-01-01T14:00:00Z, usageToken: '...' } }

📥 Agent 请求 100 API 配额:
{ status: 'approved', allocation: { requestId: 'req_yyy', resourceType: 'api_quota', resourceId: ['resource_api_quota_0', ...], expiresAt: 2024-01-01T14:00:00Z, usageToken: '...' } }

📊 GPU 池状态: { type: 'gpu', total: 4, available: 2, allocated: 2, queueLength: 0 }


### 4.4 核心模式解析

**场景三的核心模式**：`resource_request` + `info_request`

1. **所有资源请求都经过协调中心**：没有直接抢占，每个请求都有记录
2. **优先级队列**：高优先级任务排在前面，确保关键任务优先
3. **自动释放和通知**：资源用完后自动进入队列下一个
4. **Token 验证**：每个分配都有 Token，防止冒用

### 4.5 关键注意事项

1. **不要让 Agent 直接访问资源池**：统一通过协调中心分配，避免竞态条件
2. **资源必须有 TTL**：设置过期时间，确保资源不会永久占用
3. **队列要有优先级**：否则低优先级任务可能永远等不到资源
4. **要有审计日志**：记录谁在什么时间申请了什么资源

---

## 五、场景四：长流程状态追踪

### 5.1 业务背景

有些业务流程很复杂，不是一个 Agent 能完成的——需要多个 Agent 协作，涉及多轮消息交互，跨越数小时甚至数天。

比如：一个「用户数据导入」流水线：
- 步骤 1：验证文件格式（Fast）
- 步骤 2：清洗和转换数据（慢，5-30 分钟）
- 步骤 3：导入数据库（中等，1-5 分钟）
- 步骤 4：发送通知邮件（Fast）

每个步骤可能由不同的 Agent 执行，步骤之间有依赖关系。如果中间某步失败了，需要能够：
1. 知道当前在哪一步
2. 重试失败步骤
3. 恢复到某个已知状态

这需要一套完整的状态机 + A2A 状态同步机制。

plaintext
┌──────────────────────────────────────────────────────────────┐
│ 状态机 Agent │
│ (Pipeline Orchestrator) │
└────────────────────────┬─────────────────────────────────────┘
│
┌───────────────────┼───────────────────┐
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Validator │───▶│ Cleanser │───▶│ Importer │───▶ Done
│ Agent │ │ Agent │ │ Agent │
└────────────┘ └────────────┘ └────────────┘
│ │ │
└──────────────────┴──────────────────┘
│
status_sync
(上报状态)


### 5.2 完整代码

typescript
// 文件：a2a_pipeline_orchestrator.ts

import {
spawn_subagent,
agent_send_message,
delegate_message
} from '@markus/agent';

// ========== 类型定义 ==========

interface PipelineStatus {
pipelineId: string;
currentStage: PipelineStage;
progress: number; // 0-100
stageDetails: Record;
startedAt: string;
updatedAt: string;
errors: PipelineError[];
}

interface StageDetail {
stage: PipelineStage;
status: 'pending' | 'running' | 'completed' | 'failed';
startedAt?: string;
completedAt?: string;
recordsProcessed?: number;
recordsFailed?: number;
}

interface PipelineError {
stage: PipelineStage;
timestamp: string;
message: string;
recoverable: boolean;
}

interface StatusSubscriber {
agentId: string;
stages: PipelineStage[]; // 订阅哪些阶段的状态
}

// 流水线编排器
class PipelineOrchestrator {
private pipelines: Map = new Map();
private subscribers: StatusSubscriber[] = [];
private agents: Map = new Map(); // stage -> agentId

constructor() {
this.setupSubscribers();
console.log('✅ Pipeline Orchestrator initialized');
}

private setupSubscribers() {
// 设置默认订阅者
// Dashboard 订阅所有阶段
// Alerting 只订阅失败状态
// Audit Logger 订阅所有阶段
console.log('📡 Default subscribers configured');
}

// 启动新流水线
async startPipeline(dataSource: string): Promise {
const pipelineId = pipeline_${Date.now()};

const status: PipelineStatus = {
  pipelineId,
  currentStage: 'ingestion',
  progress: 0,
  stageDetails: {
    ingestion: { stage: 'ingestion', status: 'pending' },
    validation: { stage: 'validation', status: 'pending' },
    transformation: { stage: 'transformation', status: 'pending' },
    storage: { stage: 'storage', status: 'pending' }
  },
  startedAt: new Date().toISOString(),
  updatedAt: new Date().toISOString(),
  errors: []
};

this.pipelines.set(pipelineId, status);

// 启动各阶段 Agent
await this.spawnStageAgents();

// 通知 ingestion agent 开始
const ingestionId = this.agents.get('ingestion');
if (ingestionId) {
  await agent_send_message({
    agent_id: ingestionId,
    message: `start_stage:${JSON.stringify({ pipelineId, dataSource })}`
  });
}

console.log(`🚀 Pipeline ${pipelineId} started`);
return pipelineId;

}

private async spawnStageAgents() {
const stageConfigs = [
{ stage: 'ingestion', prompt: '你负责数据摄入。从指定数据源读取原始数据，返回 records。' },
{ stage: 'validation', prompt: '你负责数据验证。验证每条记录是否符合规则，返回 validCount 和 invalidCount。' },
{ stage: 'transformation', prompt: '你负责数据转换。将数据转换为目标格式并返回 transformedCount。' },
{ stage: 'storage', prompt: '你负责数据存储。将数据写入目标存储并返回 storedCount。' }
];

for (const cfg of stageConfigs) {
  const agentId = await spawn_subagent({
    system_prompt: cfg.prompt,
    allowed_tools: []
  });
  this.agents.set(cfg.stage, agentId);
  console.log(`   Spawned ${cfg.stage} agent: ${agentId}`);
}

}

// 接收 Agent 的状态报告
async handleStatusReport(agentId: string, report: StatusReport) {
const pipeline = this.findPipelineByAgent(agentId);
if (!pipeline) return;

const detail = pipeline.stageDetails[report.stage];
if (detail) {
  detail.status = report.status;
  if (report.startedAt) detail.startedAt = report.startedAt;
  if (report.completedAt) {
    detail.completedAt = report.completedAt;
    detail.recordsProcessed = report.recordsProcessed;
    detail.recordsFailed = report.recordsFailed;
  }
}

pipeline.updatedAt = new Date().toISOString();
pipeline.progress = this.calculateProgress(pipeline);
pipeline.currentStage = this.determineCurrentStage(pipeline);

if (report.status === 'failed') {
  pipeline.errors.push({
    stage: report.stage,
    timestamp: new Date().toISOString(),
    message: report.message || 'Unknown error',
    recoverable: report.recoverable || false
  });
}

this.broadcastStatus(pipeline);

// 如果当前阶段完成，触发下一阶段
if (report.status === 'completed') {
  await this.triggerNextStage(pipeline, report.stage);
}

}

private findPipelineByAgent(agentId: string): PipelineStatus | null {
for (const pipeline of this.pipelines.values()) {
if (pipeline.currentStage === agentId) return pipeline;
}
return null;
}

private calculateProgress(pipeline: PipelineStatus): number {
const stageWeights: Record = {
ingestion: 15,
validation: 25,
transformation: 40,
storage: 20
};

let total = 0;
for (const [stage, detail] of Object.entries(pipeline.stageDetails)) {
  if (detail.status === 'completed') {
    total += stageWeights[stage] || 25;
  } else if (detail.status === 'running') {
    total += (stageWeights[stage] || 25) * 0.5;
  }
}
return Math.round(total);

}

private determineCurrentStage(pipeline: PipelineStatus): PipelineStage {
for (const [stage, detail] of Object.entries(pipeline.stageDetails)) {
if (detail.status === 'pending' || detail.status === 'running') {
return stage as PipelineStage;
}
}
return 'completed';
}

private async triggerNextStage(pipeline: PipelineStatus, completedStage: string) {
const stageOrder: PipelineStage[] = ['ingestion', 'validation', 'transformation', 'storage'];
const nextIdx = stageOrder.indexOf(completedStage) + 1;

if (nextIdx < stageOrder.length) {
  const nextStage = stageOrder[nextIdx];
  const nextAgentId = this.agents.get(nextStage);
  if (nextAgentId) {
    const detail = pipeline.stageDetails[nextStage];
    detail.status = 'running';
    detail.startedAt = new Date().toISOString();

    await agent_send_message({
      agent_id: nextAgentId,
      message: `start_stage:${JSON.stringify({ pipelineId: pipeline.pipelineId })}`
    });

    console.log(`   → Triggered next stage: ${nextStage}`);
  }
} else {
  pipeline.currentStage = 'completed';
  pipeline.progress = 100;
  console.log('✅ Pipeline completed');
}

}

private broadcastStatus(pipeline: PipelineStatus) {
for (const sub of this.subscribers) {
const interestedStages = sub.stages;
if (interestedStages.includes(pipeline.currentStage) || interestedStages.includes('all')) {
console.log(📡 Broadcasting to ${sub.agentId}: ${pipeline.currentStage});
}
}
}
}

// ========== 使用示例 ==========

async function demonstratePipelineOrchestration() {
const orchestrator = new PipelineOrchestrator();

// 启动流水线
const pipelineId = await orchestrator.startPipeline('s3://data/users.csv');

// 模拟各阶段完成
console.log('\n⏳ 模拟流水线执行...');

// 阶段 1 完成
await new Promise(r => setTimeout(r, 100));
await orchestrator.handleStatusReport('agent_ingestion', {
pipelineId,
stage: 'ingestion',
status: 'completed',
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
recordsProcessed: 10000,
recordsFailed: 0
});

// 阶段 2 完成
await new Promise(r => setTimeout(r, 100));
await orchestrator.handleStatusReport('agent_validation', {
pipelineId,
stage: 'validation',
status: 'completed',
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
recordsProcessed: 9500,
recordsFailed: 500
});

// 阶段 3 完成
await new Promise(r => setTimeout(r, 100));
await orchestrator.handleStatusReport('agent_transformation', {
pipelineId,
stage: 'transformation',
status: 'completed',
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
recordsProcessed: 9500,
recordsFailed: 0
});

// 阶段 4 完成
await new Promise(r => setTimeout(r, 100));
await orchestrator.handleStatusReport('agent_storage', {
pipelineId,
stage: 'storage',
status: 'completed',
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
recordsProcessed: 9500,
recordsFailed: 0
});

return pipelineId;
}

demonstratePipelineOrchestration();


运行效果：

plaintext
🚀 Pipeline pipeline_1710500000000 started
Spawned ingestion agent: agent_001
Spawned validation agent: agent_002
Spawned transformation agent: agent_003
Spawned storage agent: agent_004

⏳ 模拟流水线执行...
📊 [pipeline_1710500000000] Stage: ingestion, Progress: 15%
[pipeline_1710500000000] 📥 Ingesting data
✅ Stage ingestion completed: 10000 records

📊 [pipeline_1710500000000] Stage: validation, Progress: 25%
[pipeline_1710500000000] ✅ Validating data
✅ Stage validation completed: 9500 records, 500 failed

📊 [pipeline_1710500000000] Stage: transformation, Progress: 65%
[pipeline_1710500000000] 🔄 Transforming data
✅ Stage transformation completed: 9500 records

📊 [pipeline_1710500000000] Stage: storage, Progress: 85%
[pipeline_1710500000000] 💾 Storing data
✅ Stage storage completed: 9500 records

📊 [pipeline_1710500000000] Stage: completed, Progress: 100%
✅ Pipeline completed: pipeline_1710500000000


### 5.5 关键设计模式

| 模式 | 在代码中的体现 | 作用 |
|------|--------------|------|
| **状态广播** | `broadcastStatus()` | 所有订阅者实时感知流水线状态 |
| **阶段订阅** | `subscribeToPipeline(stages)` | 按需订阅，只关心关心的阶段 |
| **失败恢复** | `handleStatusReport` 中的 `recoverable` | 可恢复错误自动重试 |
| **进度计算** | `calculateProgress()` | 实时计算总体进度 |

### 5.6 场景四的注意事项

1. **不要让 Agent 直接更新状态**：统一通过 PipelineOrchestrator，避免状态不一致
2. **失败要有恢复策略**：区分可恢复和不可恢复错误
3. **订阅者要能动态注册/取消**：不要硬编码订阅者列表
4. **日志要持久化**：流水线状态要落盘，重启后能恢复

---

## 六、总结与展望

### 6.1 四个场景回顾

| 场景 | 核心问题 | 解决方案 | A2A 消息类型 |
|------|---------|---------|------------|
| 代码审查 | 单个 Agent 能力有限 | 多 Agent 并发审查 → 汇总报告 | task_delegate → result_report |
| 资源协调 | 资源竞争和抢占 | 统一协调中心 → 优先级队列 | resource_request → info_request |
| 长流程追踪 | 状态同步困难 | Pipeline Orchestrator → 状态广播 | status_update → status_sync |

### 6.2 核心模式总结

四个场景中贯穿着一些共同模式：

1. **中心化协调**：不是让 Agent 自由协作，而是有一个中心节点负责分配任务、管理资源、追踪状态
2. **状态可见**：所有关键状态都实时广播给订阅者
3. **优雅降级**：每个场景都有错误处理和降级策略
4. **边界清晰**：每个 Agent 有明确的职责边界，不越界

### 6.3 你现在可以做什么


学完这个系列，你应该能够：


- [ ] 构建一个多 Agent 协作系统
- [ ] 使用 A2A 协议进行 Agent 间通信
- [ ] 实现任务委派和结果汇总
- [ ] 实现资源管理和调度
- [ ] 实现长流程状态追踪

### 6.4 下一步

如果你想继续深入，可以尝试：

1. **把示例代码跑起来**：用 Markus 平台部署你第一个多 Agent 系统
2. **修改场景参数**：把代码审查改成 API 测试，把资源池改成数据库连接池
3. **组合多个模式**：比如在代码审查流水线中加入 GPU 资源调度
4. **加入监控和告警**：让 Dashboard 实时显示所有流水线状态


### 6.5 A2A 协议系列总结

| 篇文章 | 主题 | 核心收获 |
|--------|------|---------|
| s01 | A2A 协议入门 | 理解为什么 Agent 需要通信 |
| s02 | 消息类型体系 | 掌握四种核心消息类型 |
| s03 | 任务委派 | 学会如何委派任务给专项 Agent |
| s04 | 协作会话 | 学会管理多轮对话 |
| s05 | 实战案例 | 用四个真实场景串联所有知识 |


**最终你会发现**：A2A 协议不只是让 Agent 发消息，而是让你能够构建**真正有用的多 Agent 系统**。

---

## 七、常见陷阱与避坑指南

A2A 协议虽然强大，但使用不当也会带来问题。这里总结了一些常见陷阱和应对方法。

### 7.1 消息风暴

**陷阱**：当多个 Agent 互相发送状态更新通知时，可能产生指数级消息量。

**场景**：100 个 Worker Agent 同时向 Supervisor 发送 `status_sync`，Supervisor 又向 100 个订阅者广播，导致 O(n²) 的消息量。

**应对**：

typescript
// 1. 批量聚合：不是每条更新都立即发送
class StatusAggregator {
private buffer: Map = new Map();
private flushInterval = 5000; // 5秒批量一次

add(update: StatusUpdate) {
const key = update.agentId;
this.buffer.set(key, [...(this.buffer.get(key) || []), update]);
}

async flush() {
// 批量发送
for (const [agentId, updates] of this.buffer) {
await agent_send_message({
to: agentId,
type: 'status_sync_batch',
payload: { updates }
});
}
this.buffer.clear();
}
}

// 2. 差异化订阅：不是所有 Agent 都订阅所有消息
// 精细化订阅策略，减少不必要的消息传递


### 7.2 死锁和活锁

**陷阱**：Agent A 等 Agent B 的消息，Agent B 等 Agent A 的消息，形成死锁。

**场景**：

plaintext
Agent A → Agent B: 需要你的结果
Agent B → Agent A: 需要你的结果


两个 Agent 互相等待，形成死锁。

**应对**：

typescript
// 1. 超时机制：设置消息超时，超时后自动释放
const result = await Promise.race([
agent_send_message({ to: 'Agent B', message: 'need_result' }),
new Promise((_, reject) => setTimeout(() => reject(new Error('timeout')), 30000))
]);

// 2. 避免循环依赖：设计时检查 Agent 之间的依赖关系
// 3. 使用中间件：所有消息经过中间件，中间件检测循环依赖


### 7.3 状态不一致

**陷阱**：多个 Agent 同时修改同一个状态，导致状态不一致。

**场景**：
- Agent A 读取状态 S = 0
- Agent B 读取状态 S = 0
- Agent A 修改 S = 1
- Agent B 修改 S = 2
- 最终 S = 2，但 Agent A 以为自己改的是 S = 1

**应对**：

typescript
// 使用乐观锁或悲观锁
class SharedStateManager {
private locks: Map> = new Map();

async withLock(key: string, fn: () => Promise) {
while (this.locks.has(key)) {
await this.locks.get(key);
}

let release: () => void;
const lock = new Promise<void>(resolve => { release = resolve; });
this.locks.set(key, lock);

try {
  await fn();
} finally {
  this.locks.delete(key);
  release();
}

}
}


### 7.4 消息丢失

**陷阱**：网络故障或 Agent 重启导致消息丢失。

**场景**：Agent A 发送消息给 Agent B，但 Agent B 在消息到达前重启，导致消息丢失。

**应对**：

typescript
// 1. 消息持久化：所有消息先落盘再发送
// 2. 确认机制：接收方收到消息后返回确认
// 3. 重试机制：发送方在超时后重试发送

interface MessageWithAck {
id: string;
payload: A2AMessage;
acked: boolean;
retryCount: number;
}

class ReliableMessenger {
private pending: Map = new Map();

async send(message: A2AMessage): Promise {
const id = generateMessageId();
this.pending.set(id, {
id,
payload: message,
acked: false,
retryCount: 0
});

await this.persistToDisk(id, message);
await this.trySend(id);

}

private async trySend(id: string) {
const msg = this.pending.get(id);
if (!msg || msg.acked) return;

try {
  await agent_send_message(msg.payload);
  await this.waitForAck(id, 5000); // 等待 5 秒确认
} catch (e) {
  msg.retryCount++;
  if (msg.retryCount < 3) {
    setTimeout(() => this.trySend(id), 1000 * msg.retryCount);
  }
}

}
}


### 7.5 优先级反转

**陷阱**：高优先级任务等待低优先级任务释放资源。

**场景**：
1. 低优先级任务占用了 GPU
2. 高优先级任务请求 GPU
3. 高优先级任务等待低优先级任务
4. 但低优先级任务正在等待高优先级任务的输出

**死锁！**

**应对**：

typescript
// 优先级继承：高优先级等待时，提升低优先级任务的优先级
class PriorityInheritanceScheduler {
async requestResource(agentId: string, resource: string, priority: number) {
const holder = this.getResourceHolder(resource);

if (holder && priority > holder.priority) {
  // 提升持有者的优先级
  holder.priority = priority;
  console.log(`🔼 提升 ${holder.agentId} 的优先级到 ${priority}`);
}

return this.doAllocate(agentId, resource, priority);

}
}


---

## 八、性能优化建议

### 8.1 消息压缩

对于大量文本消息，可以使用压缩算法减少带宽：

typescript
import { gzip, gunzip } from 'zlib';

async function compressMessage(message: A2AMessage): Promise {
const json = JSON.stringify(message);
return new Promise((resolve, reject) => {
gzip(json, (err, result) => {
if (err) reject(err);
else resolve(result);
});
});
}


### 8.2 消息批处理

对于大量小消息，批处理可以显著减少网络开销：

typescript
class MessageBatcher {
private buffer: A2AMessage[] = [];
private flushInterval = 100; // 100ms

constructor() {
setInterval(() => this.flush(), this.flushInterval);
}

add(message: A2AMessage) {
this.buffer.push(message);
}

private async flush() {
if (this.buffer.length === 0) return;

const batch = [...this.buffer];
this.buffer = [];

await agent_send_message({
  type: 'batch',
  payload: { messages: batch }
});

}
}


### 8.3 缓存热点数据

对于频繁访问的数据，使用缓存减少重复计算：

typescript
class AgentCache {
private cache: Map = new Map();

get(key: string): any | null {
const entry = this.cache.get(key);
if (!entry) return null;
if (Date.now() > entry.expiry) {
this.cache.delete(key);
return null;
}
return entry.data;
}

set(key: string, data: any, ttlMs: number = 60000) {
this.cache.set(key, {
data,
expiry: Date.now() + ttlMs
});
}
}




---

## 九、最佳实践清单

### 9.1 设计阶段

- [ ] 明确每个 Agent 的职责边界
- [ ] 绘制 Agent 关系图和消息流图
- [ ] 定义消息类型和格式（使用 TypeScript interface）
- [ ] 设计错误处理和降级策略
- [ ] 确定监控和告警方案

### 9.2 实现阶段

- [ ] 使用 TypeScript 严格模式，确保类型安全
- [ ] 每个消息都有唯一 ID，方便追踪
- [ ] 消息发送前做格式验证
- [ ] 实现超时和重试机制
- [ ] 日志记录每个关键节点

### 9.3 测试阶段

- [ ] 单元测试：每个 Agent 独立测试
- [ ] 集成测试：Agent 之间消息交互测试
- [ ] 压力测试：大量并发消息下的表现
- [ ] 故障注入测试：模拟网络故障、Agent 重启

### 9.4 监控阶段

- [ ] 部署 Dashboard 实时显示系统状态
- [ ] 设置告警阈值（超时数、失败率、队列长度）
- [ ] 定期分析消息日志，优化性能
- [ ] 记录每个 Agent 的资源消耗

---

## 结语

A2A 协议是构建复杂多 Agent 系统的基石。通过这四个实战场景，你应该已经掌握了：

1. **任务分解与委托**：如何让主 Agent 协调多个专项 Agent
2. **并发与协调**：如何处理多个 Agent 的并发执行和结果汇总
3. **资源管理**：如何实现智能的资源分配和调度
4. **状态追踪**：如何实现长流程的实时状态同步

更重要的是，你学会了**避坑**——知道什么不该做，比知道什么该做更重要。

A2A 协议的力量在于**组合**。你可以把今天学的模式组合起来，构建更复杂的系统：
- 研究主管 + 多个专项 Agent + 资源协调中心 + Dashboard
- 代码审查流水线 + 安全审查 + 资源池管理 + 审计日志

关键是：**从简单开始，逐步演进**。不要一开始就设计完美的架构，先跑起来，再优化。

祝你构建出强大的多 Agent 系统！

---

*本文是 A2A Protocol Series 的最后一篇（s05），也是本系列的最后一篇。*


*系列文章：*
- *s01: [A2A 协议入门：为什么 Agent 需要通信](#)*
- *s02: [消息类型体系：四种核心消息](#)*
- *s03: [任务委派：从一个 Agent 到多个 Agent](#)*
- *s04: [协作会话：多轮对话的艺术](#)*
- *s05: [实战案例：四个真实协作场景](#)（本文）*

Markus A2A Protocol in Action: Four Real-World Collaboration Scenarios

Jason — Thu, 16 Apr 2026 11:29:26 +0000

Introduction: Theory Is Not Enough

In our previous four articles (s01-s04), we've covered the A2A protocol fundamentals:

s01 explained why agents need communication—single agents can't handle complex tasks, collaboration is the way
s02 covered the message type system—task_delegate, status_sync, resource_request, info_request, each with its own purpose
s03 explained task delegation mechanisms—how to split large tasks into subtasks and distribute them across agents
s04 covered collaborative sessions and session management—how to organize multi-turn conversations and track them

But knowing theory doesn't mean you can use it. Like reading a swimming manual doesn't guarantee you can swim—the real learning happens in the water.

This article (s05) is learning to swim in the water. We won't discuss protocol details; instead, we'll look at four real use cases, each with complete code examples that can run in production.

Scenario 1: Research Task Decomposition

Background: The marketing team needs a comprehensive competitive analysis report. The Research Director receives the task and needs to coordinate three specialized researchers (Market Researcher, Technical Analyst, Financial Analyst) to work in parallel.

Step 1: Task Analysis and Decomposition

// Research Director analyzes the task and creates a decomposition plan
interface TaskDecomposition {
  mainTaskId: string;
  subtasks: SubTaskPlan[];
  dependencies: DependencyGraph;
  resultAggregator: string; // Agent ID responsible for aggregating results
}

const decomposition: TaskDecomposition = {
  mainTaskId: 'research_competitive_2024Q1',
  subtasks: [
    {
      id: 'sub_001',
      title: "'Market Research - Competitor Landscape',"
      agentType: 'market-researcher',
      description: "'Research top 5 competitors: product offerings, market share, user base, growth trajectory',"
      priority: 'high',
      estimatedDuration: 30, // minutes
      outputFormat: 'structured_report'
    },
    {
      id: 'sub_002',
      title: "'Technical Analysis - Technology Stack',"
      agentType: 'technical-analyst',
      description: "'Analyze competitors technical stack: architecture, scalability, tech innovation',"
      priority: 'high',
      estimatedDuration: 25,
      outputFormat: 'technical_report'
    },
    {
      id: 'sub_003',
      title: "'Financial Analysis - Business Model',"
      agentType: 'financial-analyst',
      description: "'Research competitors financials: revenue, funding, unit economics, pricing strategy',"
      priority: 'medium',
      estimatedDuration: 35,
      outputFormat: 'financial_report'
    }
  ],
  dependencies: {
    'sub_001': [],
    'sub_002': [],
    'sub_003': [],
    // Results aggregation starts only after all three complete
    'aggregation': ['sub_001', 'sub_002', 'sub_003']
  },
  resultAggregator: 'research-director'
};

Step 2: Subtask Delegation via A2A

// Research Director delegates tasks to three specialized agents
async function delegateResearchTasks(decomposition: TaskDecomposition) {
  const results: Map<string, any> = new Map();
  const agentIds = await discoverAgentsByType([
    'market-researcher',
    'technical-analyst', 
    'financial-analyst'
  ]);

  // Send delegation messages to all agents in parallel
  const delegationPromises = decomposition.subtasks.map(async (subtask, index) => {
    const agentId = agentIds[index];

    const message: A2AMessage = {
      messageId: generateMessageId(),
      type: 'task_delegate',
      from: 'research-director',
      to: agentId,
      timestamp: new Date().toISOString(),
      payload: {
        taskId: subtask.id,
        title: "subtask.title,"
        description: "subtask.description,"
        priority: subtask.priority,
        deadline: calculateDeadline(subtask.estimatedDuration),
        outputFormat: subtask.outputFormat,
        callbackChannel: 'research-director' // Where to send results
      },
      headers: {
        correlationId: decomposition.mainTaskId,
        traceId: generateTraceId()
      }
    };

    await agent_send_message({
      agent_id: agentId,
      message: JSON.stringify(message),
      wait_for_reply: false
    });

    console.log(`[Delegated] ${subtask.id} → ${agentId}`);
    return subtask.id;
  });

  await Promise.all(delegationPromises);
  console.log(`All ${decomposition.subtasks.length} tasks delegated`);

  return results;
}

Step 3: Parallel Execution and Status Monitoring

// Monitor execution progress and collect results
class ResearchProgressMonitor {
  private results: Map<string, any> = new Map();
  private status: Map<string, 'pending' | 'in_progress' | 'completed' | 'failed'> = new Map();

  async start() {
    // Subscribe to status updates from all agents
    for (const subtask of decomposition.subtasks) {
      this.status.set(subtask.id, 'pending');

      // Simulate status polling (in production, use event-based notifications)
      this.pollStatus(subtask.id, 10000); // Check every 10 seconds
    }
  }

  async pollStatus(taskId: string, intervalMs: number) {
    setInterval(async () => {
      const status = await queryAgentStatus(taskId);
      this.status.set(taskId, status.state);

      if (status.state === 'completed') {
        this.results.set(taskId, status.result);
        this.onSubtaskComplete(taskId, status.result);
      }

      if (status.state === 'failed') {
        this.onSubtaskFailed(taskId, status.error);
      }
    }, intervalMs);
  }

  onSubtaskComplete(taskId: string, result: any) {
    console.log(`[Complete] ${taskId} received result of ${JSON.stringify(result).length} chars`);

    // Check if all subtasks are done
    const allDone = Array.from(this.status.values())
      .every(s => s === 'completed' || s === 'failed');

    if (allDone) {
      this.aggregateResults();
    }
  }

  async aggregateResults() {
    const aggregated = {
      timestamp: new Date().toISOString(),
      mainTaskId: decomposition.mainTaskId,
      results: Object.fromEntries(this.results),
      summary: this.generateSummary()
    };

    // Send aggregated report to user
    await notify_user({
      title: "'Competitive Analysis Report Ready',"
      body: JSON.stringify(aggregated, null, 2)
    });
  }
}

Step 4: Results Aggregation and Report Generation

// Aggregate all research results into a comprehensive report
async function aggregateResearchResults(results: Map<string, any>) {
  const marketReport = results.get('sub_001');
  const technicalReport = results.get('sub_002');
  const financialReport = results.get('sub_003');

  const finalReport = {
    title: "'Competitive Analysis Report - Q1 2024',"
    executiveSummary: generateExecutiveSummary(marketReport, technicalReport, financialReport),
    marketAnalysis: marketReport.content,
    technicalAnalysis: technicalReport.content,
    financialAnalysis: financialReport.content,
    recommendations: generateRecommendations(marketReport, technicalReport, financialReport),
    appendices: {
      dataSources: collectDataSources([marketReport, technicalReport, financialReport]),
      methodology: 'Multi-agent parallel research with expert synthesis'
    }
  };

  return finalReport;
}

Key Patterns in This Scenario:

Task decomposition: Large tasks split into parallel subtasks
Agent discovery: Find right agents by type/capability
Parallel delegation: Send tasks to multiple agents simultaneously
Status monitoring: Track progress of all parallel tasks
Result aggregation: Combine outputs from different agents
Deadline management: Set reasonable time limits for each subtask

Scenario 2: Code Review Pipeline

Background: A pull request with 15 files was submitted. We need to run three types of reviews (code quality, security, performance) in parallel, then aggregate results into a final review report.

Step 1: PR Analysis and Review Task Planning

interface ReviewTask {
  agentId: string;
  files: PRFile[];
  focus: string;  // e.g., 'code-quality', 'security', 'performance'
}

interface ReviewerConfig {
  name: string;
  focusPatterns: RegExp[];
  priority: number;
}

// Three types of reviewers
const reviewerConfigs: ReviewerConfig[] = [
  {
    name: 'code-quality-reviewer',
    focusPatterns: [/\.ts$/, /\.js$/, /\.tsx$/],
    priority: 1
  },
  {
    name: 'security-reviewer',
    focusPatterns: [/auth/, /security/, /permission/, /\.ts$/],
    priority: 2
  },
  {
    name: 'performance-reviewer',
    focusPatterns: [/database/, /api/, /cache/, /query/],
    priority: 3
  }
];

// Analyze PR files and distribute to appropriate reviewers
function prepareReviewTasks(files: PRFile[], configs: typeof reviewerConfigs): ReviewTask[] {
  const tasks: ReviewTask[] = [];

  for (const config of configs) {
    const matchedFiles = files.filter(f => 
      config.focusPatterns.some(pattern => pattern.test(f.path))
    );

    if (matchedFiles.length > 0) {
      tasks.push({
        agentId: '', // Will be resolved by agent registry
        files: matchedFiles,
        focus: config.name.replace('-reviewer', '').replace('-', ' ').toUpperCase()
      });
    }
  }

  return tasks;
}

// Example: 15 files in PR
const prFiles: PRFile[] = [
  { path: 'src/auth/login.ts', additions: 150, deletions: 20 },
  { path: 'src/auth/permissions.ts', additions: 80, deletions: 10 },
  { path: 'src/api/users.ts', additions: 200, deletions: 30 },
  { path: 'src/database/query.ts', additions: 120, deletions: 15 },
  { path: 'src/cache/redis.ts', additions: 60, deletions: 5 },
  // ... more files
];

const reviewTasks = prepareReviewTasks(prFiles, reviewerConfigs);
// Result: code quality → 12 files, security → 3 files, performance → 5 files

Step 2: Concurrent Review Execution

// Run all three reviews concurrently
async function runConcurrentReviews(reviewTasks: ReviewTask[]) {
  // Find available agents for each review type
  const agents = await Promise.all(
    reviewTasks.map(task => discoverAgentByFocus(task.focus))
  );

  // Create review delegation messages
  const reviewMessages = reviewTasks.map((task, i) => ({
    to: agents[i],
    message: {
      type: 'task_delegate',
      payload: {
        taskId: `review_${task.focus.toLowerCase()}_${Date.now()}`,
        title: `${task.focus} Code Review`,
        files: task.files.map(f => f.path),
        reviewCriteria: getReviewCriteria(task.focus),
        outputFormat: 'structured_issues'
      }
    }
  }));

  // Send all messages in parallel
  const results = await Promise.allSettled(
    reviewMessages.map(({ to, message }) => 
      agent_send_message({ agent_id: to, message: JSON.stringify(message) })
    )
  );

  // Handle any failures
  results.forEach((result, i) => {
    if (result.status === 'rejected') {
      console.error(`Review ${i} failed:`, result.reason);
    }
  });

  return results;
}

Step 3: Aggregating Reviews into Final Report

// Aggregate review results from all three reviewers
interface ReviewResult {
  focus: string;
  issues: Issue[];
  score: number; // 0-10
  recommendations: string[];
}

interface AggregatedReview {
  prId: string;
  overallScore: number;
  criticalIssues: Issue[];
  warnings: Issue[];
  suggestions: Issue[];
  reviewDetails: ReviewResult[];
  consensus: string; // 'approve' | 'request_changes' | 'reject'
}

function aggregateReviewResults(results: ReviewResult[]): AggregatedReview {
  // Group issues by severity
  const allIssues = results.flatMap(r => r.issues);
  const criticalIssues = allIssues.filter(i => i.severity === 'critical');
  const warnings = allIssues.filter(i => i.severity === 'warning');
  const suggestions = allIssues.filter(i => i.severity === 'suggestion');

  // Calculate overall score (weighted average)
  const weights = { critical: 0.5, warning: 0.3, suggestion: 0.2 };
  const overallScore = results.reduce((sum, r) => sum + r.score, 0) / results.length;

  // Determine consensus
  let consensus: 'approve' | 'request_changes' | 'reject';
  if (criticalIssues.length > 0) {
    consensus = 'reject';
  } else if (warnings.length > 3) {
    consensus = 'request_changes';
  } else {
    consensus = 'approve';
  }

  return {
    prId: 'PR-1234',
    overallScore,
    criticalIssues,
    warnings,
    suggestions,
    reviewDetails: results,
    consensus
  };
}

Key Patterns in This Scenario:

Smart file routing: Distribute files to right reviewers based on content type
Concurrent execution: Multiple reviewers work simultaneously
Result merging: Combine outputs with severity-based grouping
Weighted scoring: Different issue types affect final score differently
Consensus logic: Automatic approve/reject based on issue severity

Scenario 3: Cross-Team Resource Coordination

Background: A large-scale data migration requires coordinated GPU resources across three teams (ML team, Data team, Backend team). A Resource Coordinator acts as the central hub to manage allocation, scheduling, and conflict resolution.

Step 1: Resource Discovery and Registration

// Resource Coordinator manages all shared resources
class ResourceCoordinator {
  private pools: Map<string, ResourcePool> = new Map();

  constructor() {
    // Initialize resource pools for each team
    this.pools.set('gpu', {
      type: 'gpu',
      totalCapacity: 24,  // 24 GPUs total across teams
      allocated: 0,
      reservations: [],
      allocations: new Map()
    });

    this.pools.set('memory', {
      type: 'memory',
      totalCapacity: 512,  // 512 GB total
      allocated: 0,
      reservations: [],
      allocations: new Map()
    });
  }

  // Teams register their available resources
  async registerResources(team: string, resources: Resource[]) {
    for (const resource of resources) {
      const pool = this.pools.get(resource.type);
      if (pool) {
        pool.totalCapacity += resource.capacity;
        console.log(`[Registered] ${team} added ${resource.capacity} ${resource.type}`);
      }
    }
  }

  // Show current resource status
  getResourceStatus(): ResourceStatus[] {
    return Array.from(this.pools.entries()).map(([type, pool]) => ({
      type,
      total: pool.totalCapacity,
      allocated: pool.allocated,
      available: pool.totalCapacity - pool.allocated,
      utilization: `${((pool.allocated / pool.totalCapacity) * 100).toFixed(1)}%`
    }));
  }
}

Step 2: Resource Request and Allocation

interface ResourceRequest {
  requestId: string;
  teamId: string;
  resourceType: 'gpu' | 'memory';
  quantity: number;
  priority: 'critical' | 'high' | 'normal' | 'low';
  duration: number;  // Expected usage time in ms
  purpose: string;
  expiry: string;
  token?: string;  // For reservation confirmation
}

// Handle resource requests with priority-based allocation
async function handleResourceRequest(request: ResourceRequest): Promise<ResourceAllocation> {
  const coordinator = new ResourceCoordinator();
  const pool = coordinator.pools.get(request.resourceType);

  // Check if enough resources available
  if (pool.allocated + request.quantity > pool.totalCapacity) {
    // Insufficient resources - implement queuing or rejection
    return {
      success: false,
      reason: 'insufficient_resources',
      queuePosition: await queueRequest(request)
    };
  }

  // Generate reservation token
  const token = coordinator.generateToken(request.requestId);

  // Allocate resources
  pool.allocated += request.quantity;
  pool.allocations.set(request.requestId, {
    requestId: request.requestId,
    quantity: request.quantity,
    allocatedAt: new Date().toISOString(),
    expiresAt: request.expiry
  });

  return {
    success: true,
    token,
    allocatedQuantity: request.quantity,
    expiresAt: request.expiry
  };
}

Step 3: Cross-Team Coordination via A2A

// Resource Coordinator communicates with team agents
async function coordinateResources(requests: ResourceRequest[]) {
  const results: Map<string, ResourceAllocation> = new Map();

  // Process requests in priority order
  requests.sort((a, b) => {
    const priorityOrder = { critical: 0, high: 1, normal: 2, low: 3 };
    return priorityOrder[a.priority] - priorityOrder[b.priority];
  });

  for (const request of requests) {
    // Send request to resource coordinator agent
    const result = await agent_send_message({
      agent_id: 'resource-coordinator',
      message: JSON.stringify({
        type: 'resource_request',
        payload: request
      }),
      wait_for_reply: true
    });

    results.set(request.requestId, result);

    // Notify team of allocation result
    if (result.success) {
      await agent_send_message({
        agent_id: request.teamId,
        message: JSON.stringify({
          type: 'resource_allocated',
          payload: {
            token: result.token,
            quantity: result.allocatedQuantity,
            expiresAt: result.expiresAt
          }
        })
      });
    }
  }

  return results;
}

Key Patterns in This Scenario:

Central registry: One place to track all shared resources
Priority-based allocation: Critical tasks get resources first
Reservation system: Tokens ensure allocated resources are claimed
TTL management: Resources auto-release after timeout
Cross-team communication: A2A messages coordinate actions

Scenario 4: Long-Running Workflow State Tracking

Background: A compliance audit workflow spans 5 stages over 72 hours. Multiple agents need to track state, handle failures, and provide real-time visibility to stakeholders.

Step 1: Define Workflow State Machine

type WorkflowStage = 
  | 'document_collection' 
  | 'evidence_gathering' 
  | 'analysis'
  | 'review'
  | 'report_generation';

interface WorkflowState {
  workflowId: string;
  currentStage: WorkflowStage;
  stageStatus: 'pending' | 'in_progress' | 'completed' | 'failed' | 'skipped';
  progress: number;  // 0-100
  startedAt: string;
  estimatedCompletion: string;
  stageDetails: StageDetail[];
  errors: WorkflowError[];
}

interface StageDetail {
  stage: WorkflowStage;
  status: string;
  startedAt: string;
  completedAt?: string;
  agentId: string;
  output?: any;
}

Step 2: Stage Progression via A2A Messages

// Orchestrator manages workflow progression
class WorkflowOrchestrator {
  private workflows: Map<string, WorkflowState> = new Map();

  async startWorkflow(workflowId: string) {
    const state: WorkflowState = {
      workflowId,
      currentStage: 'document_collection',
      stageStatus: 'in_progress',
      progress: 0,
      startedAt: new Date().toISOString(),
      estimatedCompletion: this.calculateETA(5),
      stageDetails: [],
      errors: []
    };

    this.workflows.set(workflowId, state);

    // Start first stage
    await this.executeStage(workflowId, 'document_collection');
  }

  async executeStage(workflowId: string, stage: WorkflowStage) {
    const state = this.workflows.get(workflowId);
    const stageAgent = this.getStageAgent(stage);

    // Notify agent to start stage
    await agent_send_message({
      agent_id: stageAgent,
      message: JSON.stringify({
        type: 'task_delegate',
        payload: {
          taskId: `${workflowId}_${stage}`,
          stage,
          workflowId,
          callback: 'workflow-orchestrator'
        }
      })
    });

    // Update state
    state.stageDetails.push({
      stage,
      status: 'in_progress',
      startedAt: new Date().toISOString(),
      agentId: stageAgent
    });

    await this.broadcastStateUpdate(workflowId);
  }

  async onStageComplete(workflowId: string, stage: WorkflowStage, output: any) {
    const state = this.workflows.get(workflowId);
    const stageDetail = state.stageDetails.find(s => s.stage === stage);

    if (stageDetail) {
      stageDetail.status = 'completed';
      stageDetail.completedAt = new Date().toISOString();
      stageDetail.output = output;
    }

    // Calculate progress
    const completedStages = state.stageDetails.filter(s => s.status === 'completed').length;
    state.progress = (completedStages / 5) * 100;

    // Move to next stage
    const nextStage = this.getNextStage(stage);
    if (nextStage) {
      state.currentStage = nextStage;
      await this.executeStage(workflowId, nextStage);
    } else {
      // Workflow complete
      state.stageStatus = 'completed';
      await this.notifyCompletion(workflowId);
    }

    await this.broadcastStateUpdate(workflowId);
  }
}

Step 3: Real-Time State Visualization

// Dashboard agent shows real-time workflow status
async function displayWorkflowDashboard(workflowId: string) {
  const state = await queryWorkflowState(workflowId);

  const dashboard = {
    workflowId,
    overall: {
      progress: `${state.progress.toFixed(1)}%`,
      status: state.stageStatus,
      startedAt: state.startedAt,
      estimatedCompletion: state.estimatedCompletion
    },
    stages: state.stageDetails.map(s => ({
      name: formatStageName(s.stage),
      status: s.status,
      duration: s.completedAt 
        ? calculateDuration(s.startedAt, s.completedAt)
        : 'in progress',
      agent: s.agentId
    })),
    errors: state.errors.map(e => ({
      stage: e.stage,
      message: e.message,
      timestamp: e.timestamp
    }))
  };

  // Send to dashboard display
  await updateDashboard(dashboard);
  return dashboard;
}

Key Patterns in This Scenario:

State machine: Clear definition of workflow stages and transitions
Persistent state: State survives agent restarts
Progress tracking: Real-time progress calculation
Stage callbacks: Agents notify orchestrator on completion
Dashboard integration: Visual status updates for stakeholders

Key Takeaways

Through these four scenarios, you've learned:

Task Decomposition and Delegation: How to coordinate multiple specialized agents
Concurrency and Coordination: How to handle parallel execution and result aggregation
Resource Management: How to implement intelligent resource allocation and scheduling
State Tracking: How to achieve real-time state synchronization for long-running workflows

Most importantly, you learned how to avoid pitfalls—knowing what not to do is more important than knowing what to do.

The power of the A2A protocol lies in composition. You can combine the patterns from today to build more complex systems:

Research Director + Multiple Specialized Agents + Resource Coordinator + Dashboard
Code Review Pipeline + Security Review + Resource Pool Management + Audit Logs

The key is: Start simple, evolve incrementally. Don't design a perfect architecture from the start—get it running first, then optimize.

Build powerful multi-agent systems!

This is the final article in the A2A Protocol Series (s05).

Series Articles:

s01: A2A Protocol Introduction: Why Agents Need Communication
s02: Message Type System: Four Core Messages
s03: Task Delegation: From One Agent to Multiple Agents
s04: Collaborative Sessions: The Art of Multi-Turn Conversations
s05: Practical Cases: Four Real Collaboration Scenarios (This article)