feat(orchestration): DAG TaskGraph + delegate bridge + A2A bus + reflex hints - #12436
Open
georgex8001 wants to merge 1 commit into
Open
feat(orchestration): DAG TaskGraph + delegate bridge + A2A bus + reflex hints#12436georgex8001 wants to merge 1 commit into
georgex8001 wants to merge 1 commit into
Conversation
alt-glitch
added
type/feature
georgex8001
force-pushed
the
feat/multi-agent-orchestration-pr
branch
from
April 23, 2026 16:42
234dde8 to
117e288
Compare
4 tasks
georgex8001
force-pushed
the
feat/multi-agent-orchestration-pr
branch
2 times, most recently
from
April 26, 2026 19:40
cde4ffd to
8f274f0
Compare
Contributor
Author
|
Hey team, just performed a clean reset and force-push. My local tests for orchestration (tests/orchestration/test_task_graph.py) are passing with flying colors. However, I noticed the CI is currently failing on two fronts, which seem to be upstream issues: Tests: There are 50+ test errors mostly originating from tests/gateway/test_discord_*.py, test_web_server.py, etc. It looks like these failures are inherited directly from the latest main branch state (possibly recent refactors in the Discord mock). Supply Chain Audit: It fails with CRITICAL supply chain risk patterns detected, but the action bot couldn't post the details in the comments due to fork GITHUB_TOKEN being read-only. Could someone confirm if the main branch is currently unstable? And if possible, could a maintainer check the audit action logs and let me know which specific pattern in the orchestration diff triggered the false positive? I'll hold off on further changes until the baseline is green and we clarify the audit flag. Thanks! |
This was referenced Jun 8, 2026
CI failure diagnosis — confirmed upstream ✅Tested locally on both Orchestration tests (PR′s new code)Discord document tests — fail on BOTH branchesRoot cause: Other CI-reported failures — all pass
Supply chain auditThe audit flag is a false positive from the fork GITHUB_TOKEN being read-only — it cannot post the actual finding. The orchestration package has zero external dependencies (stdlib-only) so there is no new supply chain surface. Bottom lineThe PR is clean. All failures are pre-existing on |
Enhancement: pre-execution DAG validation (orphans, depth limit, blast radius)I built on top of your PR to add structured pre-execution DAG validation. What's added
New public APIfrom orchestration import validate_dag
report = validate_dag(specs)
# report.passed → bool
# report.risk_level → 'routine' | 'elevated' | 'critical'
# report.findings → list[Finding] with per-step details
# report.errors → REJECT-level findings only
# report.warnings → WARNING-level findings onlyThis is the design-time validation layer. Your TaskGraph is the runtime Files changed (vs your PR branch)
Diffdiff --git a/src/orchestration/__init__.py b/src/orchestration/__init__.py
new file mode 100644
index 0000000..bc98b46
--- /dev/null
+++ b/src/orchestration/__init__.py
@@ -0,0 +1,33 @@
+"""Hermes native multi-agent orchestration (DAG + delegation + A2A bus)."""
+
+from orchestration.agent_bus import AgentCommunicationBus, AgentMessage
+from orchestration.learning import load_recent_hints, record_failure
+from orchestration.multi_agent_orchestrator import MultiAgentOrchestrator
+from orchestration.registry import OrchestratorRegistry
+from orchestration.task_graph import TaskGraph, topo_sort, validate_dag
+from orchestration.types import (
+ Finding,
+ GraphTaskRun,
+ GraphTaskSpec,
+ TaskStatus,
+ ValidationLevel,
+ ValidationReport,
+)
+
+__all__ = [
+ "AgentCommunicationBus",
+ "AgentMessage",
+ "Finding",
+ "GraphTaskRun",
+ "GraphTaskSpec",
+ "MultiAgentOrchestrator",
+ "OrchestratorRegistry",
+ "TaskGraph",
+ "TaskStatus",
+ "ValidationLevel",
+ "ValidationReport",
+ "load_recent_hints",
+ "record_failure",
+ "topo_sort",
+ "validate_dag",
+]
diff --git a/src/orchestration/task_graph.py b/src/orchestration/task_graph.py
new file mode 100644
index 0000000..4017080
--- /dev/null
+++ b/src/orchestration/task_graph.py
@@ -0,0 +1,280 @@
+"""DAG task graph with async parallel execution and pre-flight validation."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict, deque
+from typing import Awaitable, Callable, Dict, Iterable, List, Mapping, Sequence, Tuple
+
+from orchestration.types import (
+ Finding,
+ GraphTaskRun,
+ GraphTaskSpec,
+ TaskStatus,
+ ValidationLevel,
+ ValidationReport,
+)
+
+# ── Configurable thresholds ──────────────────────────────────────────────
+_DEFAULT_MAX_DEPTH_WARN = 4 # depth > this → WARNING
+_DEFAULT_MAX_DEPTH_HARD = 8 # depth > this → REJECT
+_BLAST_RADIUS_WARN = 3 # affected ≥ this → elevated risk
+_BLAST_RADIUS_CRITICAL = 5 # affected ≥ this → critical risk
+
+
+# ── Internal validation (unchanged contract) ─────────────────────────────
+
+def _validate_specs(specs: Sequence[GraphTaskSpec]) -> Dict[str, GraphTaskSpec]:
+ by_id: Dict[str, GraphTaskSpec] = {}
+ for spec in specs:
+ if spec.task_id in by_id:
+ raise ValueError(f"duplicate task_id {spec.task_id!r}")
+ by_id[spec.task_id] = spec
+ for spec in specs:
+ for dep in spec.depends_on:
+ if dep not in by_id:
+ raise ValueError(f"unknown dependency {dep!r} for task {spec.task_id!r}")
+ if spec.task_id in spec.depends_on:
+ raise ValueError(f"self-cycle on task {spec.task_id!r}")
+ # Cycle detection (DFS)
+ WHITE, GREY, BLACK = 0, 1, 2
+ color: Dict[str, int] = {k: WHITE for k in by_id}
+
+ def visit(node: str) -> None:
+ color[node] = GREY
+ for dep in by_id[node].depends_on:
+ if color[dep] == GREY:
+ raise ValueError(f"cycle detected involving {node!r} -> {dep!r}")
+ if color[dep] == WHITE:
+ visit(dep)
+ color[node] = BLACK
+
+ for tid in by_id:
+ if color[tid] == WHITE:
+ visit(tid)
+ return by_id
+
+
+# ── Public validation API ────────────────────────────────────────────────
+
+def validate_dag(
+ specs: Sequence[GraphTaskSpec],
+ *,
+ max_depth_warn: int = _DEFAULT_MAX_DEPTH_WARN,
+ max_depth_hard: int = _DEFAULT_MAX_DEPTH_HARD,
+ blast_radius_warn: int = _BLAST_RADIUS_WARN,
+ blast_radius_critical: int = _BLAST_RADIUS_CRITICAL,
+) -> ValidationReport:
+ """Run full pre-execution DAG validation.
+
+ Returns a structured ``ValidationReport`` with per-step findings,
+ risk level, and a final ``passed`` verdict. Does NOT raise — all
+ findings are collected and returned in the report.
+
+ Checks performed (in order):
+ 1. Cycle detection (DFS) — REJECT
+ 2. Orphan detection — WARNING for nodes with zero in-degree + out-degree
+ 3. Depth limit — WARNING if max depth > ``max_depth_warn``,
+ REJECT if > ``max_depth_hard``
+ 4. Blast radius — BFS from every node to compute worst-case
+ impact; risk level is elevated/critical based on thresholds
+ """
+
+ if not specs:
+ return ValidationReport(passed=True, risk_level="routine")
+
+ findings: list[Finding] = []
+ risk_level = "routine"
+
+ # ── 1. Cycle check ───────────────────────────────────────────────
+ try:
+ by_id = _validate_specs(tuple(specs))
+ except ValueError as exc:
+ findings.append(Finding(ValidationLevel.REJECT, str(exc)))
+ return ValidationReport(
+ passed=False,
+ findings=findings,
+ risk_level="critical",
+ )
+
+ # ── 2. Orphan detection ──────────────────────────────────────────
+ in_degree: Dict[str, int] = defaultdict(int)
+ out_degree: Dict[str, int] = defaultdict(int)
+ for spec in specs:
+ out_degree[spec.task_id] = len(spec.depends_on)
+ for dep in spec.depends_on:
+ in_degree[dep] += 1
+ _ = in_degree[spec.task_id] # ensure key exists
+
+ orphans = [
+ tid for tid in by_id
+ if in_degree.get(tid, 0) == 0 and out_degree.get(tid, 0) == 0
+ ]
+ if orphans:
+ findings.append(Finding(
+ ValidationLevel.WARNING,
+ f"Orphan node(s) detected: {', '.join(sorted(orphans))}. "
+ "These have no dependencies and nothing depends on them.",
+ nodes=sorted(orphans),
+ ))
+
+ # ── 3. Depth limit ───────────────────────────────────────────────
+ # Compute depth = longest path from any root (no predecessors) to node.
+ roots = [tid for tid, spec in by_id.items() if len(spec.depends_on) == 0]
+ # Build forward adjacency: who directly depends on me?
+ children: Dict[str, list[str]] = defaultdict(list)
+ for spec in specs:
+ for dep in spec.depends_on:
+ children[dep].append(spec.task_id)
+
+ depth: Dict[str, int] = {}
+ q: deque[tuple[str, int]] = deque((r, 0) for r in roots)
+ while q:
+ node, d = q.popleft()
+ if d > depth.get(node, -1):
+ depth[node] = d
+ for child in children.get(node, []):
+ q.append((child, d + 1))
+
+ max_depth = max(depth.values()) if depth else 0
+ deep_nodes = [tid for tid, d in depth.items() if d > max_depth_warn]
+
+ if max_depth > max_depth_hard:
+ findings.append(Finding(
+ ValidationLevel.REJECT,
+ f"DAG max depth {max_depth} exceeds hard limit {max_depth_hard}. "
+ f"Deepest nodes: {', '.join(sorted(deep_nodes, key=lambda n: -depth[n])[:5])}.",
+ nodes=sorted(deep_nodes),
+ ))
+ risk_level = "critical"
+ elif max_depth > max_depth_warn:
+ findings.append(Finding(
+ ValidationLevel.WARNING,
+ f"DAG max depth {max_depth} exceeds recommended limit {max_depth_warn}. "
+ f"Deepest nodes: {', '.join(sorted(deep_nodes, key=lambda n: -depth[n])[:5])}.",
+ nodes=sorted(deep_nodes),
+ ))
+ if risk_level == "routine":
+ risk_level = "elevated"
+
+ # ── 4. Blast radius ──────────────────────────────────────────────
+ # Reuse children map from depth computation above.
+
+ max_affected = 0
+ worst_node = ""
+ for tid in by_id:
+ visited: set[str] = {tid}
+ bfs_q: deque[str] = deque([tid])
+ while bfs_q:
+ cur = bfs_q.popleft()
+ for child in children.get(cur, []):
+ if child not in visited:
+ visited.add(child)
+ bfs_q.append(child)
+ affected = len(visited) - 1 # exclude self
+ if affected > max_affected:
+ max_affected = affected
+ worst_node = tid
+
+ if max_affected >= blast_radius_critical:
+ findings.append(Finding(
+ ValidationLevel.WARNING,
+ f"Critical blast radius: failure of '{worst_node}' would affect "
+ f"{max_affected} downstream node(s).",
+ nodes=[worst_node],
+ ))
+ if risk_level == "routine":
+ risk_level = "critical"
+ elif max_affected >= blast_radius_warn:
+ findings.append(Finding(
+ ValidationLevel.WARNING,
+ f"Elevated blast radius: failure of '{worst_node}' would affect "
+ f"{max_affected} downstream node(s).",
+ nodes=[worst_node],
+ ))
+ if risk_level == "routine":
+ risk_level = "elevated"
+
+ # ── Final verdict ─────────────────────────────────────────────────
+ has_reject = any(f.level == ValidationLevel.REJECT for f in findings)
+ return ValidationReport(
+ passed=not has_reject,
+ findings=findings,
+ risk_level=risk_level,
+ )
+
+
+# ── TaskGraph (runtime executor) ─────────────────────────────────────────
+
+class TaskGraph:
+ """Directed acyclic batch executor."""
+
+ def __init__(self, specs: Sequence[GraphTaskSpec]) -> None:
+ self._spec_by_id = _validate_specs(tuple(specs))
+ self._dependents: Dict[str, List[str]] = defaultdict(list)
+ self._pending_deps: Dict[str, int] = {}
+ for tid, spec in self._spec_by_id.items():
+ self._pending_deps[tid] = len(spec.depends_on)
+ for dep in spec.depends_on:
+ self._dependents[dep].append(tid)
+
+ @property
+ def task_ids(self) -> Tuple[str, ...]:
+ return tuple(self._spec_by_id.keys())
+
+ def runs(self) -> Dict[str, GraphTaskRun]:
+ return {tid: GraphTaskRun(spec=self._spec_by_id[tid]) for tid in self._spec_by_id}
+
+ async def run(
+ self,
+ execute: Callable[[GraphTaskRun], Awaitable[None]],
+ *,
+ runs: Mapping[str, GraphTaskRun] | None = None,
+ ) -> Dict[str, GraphTaskRun]:
+ """Execute nodes in parallel waves respecting dependencies."""
+
+ local: Dict[str, GraphTaskRun] = dict(runs) if runs else self.runs()
+ pending = dict(self._pending_deps)
+ queue: deque[str] = deque(tid for tid, c in pending.items() if c == 0)
+
+ while queue:
+ wave = list(queue)
+ queue.clear()
+ await asyncio.gather(*(execute(local[w]) for w in wave))
+ for tid in wave:
+ if local[tid].status == TaskStatus.FAILED:
+ continue
+ for dst in self._dependents[tid]:
+ pending[dst] -= 1
+ if pending[dst] == 0:
+ queue.append(dst)
+
+ unfinished = [tid for tid, r in local.items() if r.status == TaskStatus.PENDING]
+ if unfinished:
+ for tid in unfinished:
+ local[tid].status = TaskStatus.SKIPPED
+ local[tid].error = "skipped due to upstream failure or deadlock"
+ return local
+
+
+def topo_sort(specs: Iterable[GraphTaskSpec]) -> List[str]:
+ """Return topological order (stable tie-break by task_id)."""
+
+ specs_t = tuple(specs)
+ _validate_specs(specs_t)
+ by_id = {s.task_id: s for s in specs_t}
+ indegree: Dict[str, int] = {t: len(by_id[t].depends_on) for t in by_id}
+ ready = sorted(t for t, d in indegree.items() if d == 0)
+ order: List[str] = []
+ while ready:
+ tid = ready.pop(0)
+ order.append(tid)
+ for oid, spec in sorted(by_id.items()):
+ if tid in spec.depends_on:
+ indegree[oid] -= 1
+ if indegree[oid] == 0:
+ ready.append(oid)
+ ready.sort()
+ if len(order) != len(by_id):
+ raise ValueError("task graph has a cycle or missing nodes")
+ return order
diff --git a/src/orchestration/types.py b/src/orchestration/types.py
new file mode 100644
index 0000000..16e6bad
--- /dev/null
+++ b/src/orchestration/types.py
@@ -0,0 +1,79 @@
+"""Typed structures for DAG orchestration (stdlib only)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Mapping, MutableMapping, Sequence
+
+
+class TaskStatus(str, Enum):
+ """Lifecycle state for a graph node."""
+
+ PENDING = "pending"
+ RUNNING = "running"
+ DONE = "done"
+ FAILED = "failed"
+ SKIPPED = "skipped"
+
+
+class ValidationLevel(str, Enum):
+ """Severity of a validation finding."""
+
+ PASS = "pass"
+ WARNING = "warning"
+ REJECT = "reject"
+
+
+@dataclass(frozen=True)
+class GraphTaskSpec:
+ """DAG node definition supplied by callers."""
+
+ task_id: str
+ goal: str
+ depends_on: tuple[str, ...] = ()
+ context: str | None = None
+ toolsets: Sequence[str] | None = None
+ profile_env: Mapping[str, str] | None = None
+ metadata: Mapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class GraphTaskRun:
+ """Mutable runtime record while executing a node."""
+
+ spec: GraphTaskSpec
+ status: TaskStatus = TaskStatus.PENDING
+ summary: str | None = None
+ error: str | None = None
+ extra: MutableMapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class Finding:
+ """A single validation finding."""
+
+ level: ValidationLevel
+ message: str
+ nodes: list[str] = field(default_factory=list)
+
+
+@dataclass
+class ValidationReport:
+ """Structured result of DAG validation.
+
+ Carries per-step findings (cycle check, orphan check, depth limit,
+ blast radius) plus a summary risk_level and final verdict.
+ """
+
+ passed: bool
+ findings: list[Finding] = field(default_factory=list)
+ risk_level: str = "routine" # "routine" | "elevated" | "critical"
+
+ @property
+ def errors(self) -> list[Finding]:
+ return [f for f in self.findings if f.level == ValidationLevel.REJECT]
+
+ @property
+ def warnings(self) -> list[Finding]:
+ return [f for f in self.findings if f.level == ValidationLevel.WARNING]
diff --git a/tests/orchestration/test_task_graph.py b/tests/orchestration/test_task_graph.py
new file mode 100644
index 0000000..9ecdb88
--- /dev/null
+++ b/tests/orchestration/test_task_graph.py
@@ -0,0 +1,215 @@
+"""Unit tests for DAG orchestration primitives."""
+
+from __future__ import annotations
+
+import pytest
+
+from orchestration.task_graph import TaskGraph, topo_sort, validate_dag
+from orchestration.types import (
+ GraphTaskRun,
+ GraphTaskSpec,
+ TaskStatus,
+ ValidationLevel,
+)
+
+
+# ── Existing tests (topo sort + execution) ──────────────────────────────
+
+def test_topo_sort_chain() -> None:
+ specs = (
+ GraphTaskSpec("root", "do root"),
+ GraphTaskSpec("leaf", "do leaf", depends_on=("root",)),
+ )
+ assert topo_sort(specs) == ["root", "leaf"]
+
+
+def test_cycle_rejected() -> None:
+ specs = (
+ GraphTaskSpec("a", "ga", depends_on=("b",)),
+ GraphTaskSpec("b", "gb", depends_on=("a",)),
+ )
+ with pytest.raises(ValueError, match="cycle detected"):
+ TaskGraph(specs)
+
+
+@pytest.mark.asyncio
+async def test_fail_fast_blocks_dependents() -> None:
+ specs = (
+ GraphTaskSpec("first", "g1"),
+ GraphTaskSpec("second", "g2", depends_on=("first",)),
+ )
+ graph = TaskGraph(specs)
+ runs = graph.runs()
+
+ async def execute(run: GraphTaskRun) -> None:
+ if run.spec.task_id == "first":
+ run.status = TaskStatus.FAILED
+ run.error = "boom"
+ else:
+ run.status = TaskStatus.DONE
+
+ out = await graph.run(execute, runs=runs)
+ assert out["first"].status == TaskStatus.FAILED
+ assert out["second"].status == TaskStatus.SKIPPED
+
+
+@pytest.mark.asyncio
+async def test_parallel_wave_executes_all_roots() -> None:
+ specs = (
+ GraphTaskSpec("x", "gx"),
+ GraphTaskSpec("y", "gy"),
+ )
+ graph = TaskGraph(specs)
+ reached: list[str] = []
+
+ async def execute(run: GraphTaskRun) -> None:
+ reached.append(run.spec.task_id)
+ run.status = TaskStatus.DONE
+
+ await graph.run(execute)
+ assert set(reached) == {"x", "y"}
+
+
+# ── New validation tests ────────────────────────────────────────────────
+
+class TestValidateDag:
+ """Pre-execution DAG validation via validate_dag()."""
+
+ def test_empty_specs_passes(self) -> None:
+ report = validate_dag([])
+ assert report.passed is True
+ assert report.risk_level == "routine"
+
+ def test_valid_dag_passes(self) -> None:
+ specs = (
+ GraphTaskSpec("a", "root"),
+ GraphTaskSpec("b", "child", depends_on=("a",)),
+ )
+ report = validate_dag(specs)
+ assert report.passed is True
+ assert not report.errors
+
+ def test_cycle_rejected(self) -> None:
+ specs = (
+ GraphTaskSpec("a", "ga", depends_on=("b",)),
+ GraphTaskSpec("b", "gb", depends_on=("a",)),
+ )
+ report = validate_dag(specs)
+ assert report.passed is False
+ assert len(report.errors) == 1
+ assert "cycle" in report.errors[0].message.lower()
+ assert report.risk_level == "critical"
+
+ def test_self_cycle_rejected(self) -> None:
+ specs = (GraphTaskSpec("a", "ga", depends_on=("a",)),)
+ report = validate_dag(specs)
+ assert report.passed is False
+ assert len(report.errors) == 1
+
+ def test_unknown_dependency_rejected(self) -> None:
+ specs = (GraphTaskSpec("a", "ga", depends_on=("nonexistent",)),)
+ report = validate_dag(specs)
+ assert report.passed is False
+
+ def test_duplicate_id_rejected(self) -> None:
+ specs = (
+ GraphTaskSpec("a", "first"),
+ GraphTaskSpec("a", "second"),
+ )
+ report = validate_dag(specs)
+ assert report.passed is False
+
+ def test_orphan_node_warns(self) -> None:
+ specs = (
+ GraphTaskSpec("connected", "has edges"),
+ GraphTaskSpec("orphan", "no edges"),
+ )
+ report = validate_dag(specs)
+ assert report.passed is True # orphan is WARNING, not REJECT
+ assert len(report.warnings) >= 1
+ orphan_finding = next(
+ (f for f in report.warnings if "orphan" in f.message.lower()), None
+ )
+ assert orphan_finding is not None
+ assert "orphan" in orphan_finding.nodes
+
+ def test_deep_dag_warns(self) -> None:
+ """5 layers → max depth 4 → should WARN (default warn at >4)."""
+ specs = tuple(
+ GraphTaskSpec(f"n{i}", f"layer {i}",
+ depends_on=(f"n{i-1}",) if i > 0 else ())
+ for i in range(6) # n0 (depth 0) → n5 (depth 5)
+ )
+ report = validate_dag(specs, max_depth_warn=4)
+ depth_warnings = [
+ f for f in report.warnings if "depth" in f.message.lower()
+ ]
+ assert len(depth_warnings) >= 1
+ assert report.risk_level in ("elevated", "critical")
+
+ def test_very_deep_dag_rejected(self) -> None:
+ """10 layers → max depth 9 → should REJECT (default hard at >8)."""
+ specs = tuple(
+ GraphTaskSpec(f"n{i}", f"layer {i}",
+ depends_on=(f"n{i-1}",) if i > 0 else ())
+ for i in range(10)
+ )
+ report = validate_dag(specs)
+ assert report.passed is False
+ depth_errors = [f for f in report.errors if "depth" in f.message.lower()]
+ assert len(depth_errors) >= 1
+
+ def test_blast_radius_detected(self) -> None:
+ """Single root with 6 leaf dependents → blast radius ≥ 5 → critical."""
+ specs = [GraphTaskSpec("root", "fan-out")]
+ for i in range(6):
+ specs.append(
+ GraphTaskSpec(f"leaf{i}", "leaf", depends_on=("root",))
+ )
+ report = validate_dag(tuple(specs), blast_radius_critical=5)
+ if report.passed:
+ blast_warnings = [
+ f for f in report.warnings if "blast" in f.message.lower()
+ ]
+ assert len(blast_warnings) >= 1
+
+ def test_validate_dag_does_not_raise(self) -> None:
+ """validate_dag returns a report — it should never raise."""
+ specs = (
+ GraphTaskSpec("a", "ga", depends_on=("b",)),
+ GraphTaskSpec("b", "gb", depends_on=("a",)),
+ )
+ # Must not raise — even for cycles, it returns a report.
+ report = validate_dag(specs)
+ assert isinstance(report.passed, bool)
+
+ def test_all_checks_pass_on_clean_graph(self) -> None:
+ """A well-structured 3-node chain should pass cleanly."""
+ specs = (
+ GraphTaskSpec("orchestrator", "top-level"),
+ GraphTaskSpec("researcher", "research", depends_on=("orchestrator",)),
+ GraphTaskSpec("builder", "build", depends_on=("researcher",)),
+ )
+ report = validate_dag(specs)
+ assert report.passed is True
+ assert len(report.warnings) == 0
+ assert report.risk_level == "routine"
+
+
+class TestValidationReportProperties:
+ def test_errors_property(self) -> None:
+ specs = (
+ GraphTaskSpec("a", "ga", depends_on=("b",)),
+ GraphTaskSpec("b", "gb", depends_on=("a",)),
+ )
+ report = validate_dag(specs)
+ assert len(report.errors) >= 1
+ for f in report.errors:
+ assert f.level == ValidationLevel.REJECT
+
+ def test_warnings_property(self) -> None:
+ specs = (GraphTaskSpec("orphan", "alone"),)
+ report = validate_dag(specs)
+ assert len(report.warnings) >= 1
+ for f in report.warnings:
+ assert f.level == ValidationLevel.WARNINGBranch (if you want to pull): Happy to adjust scope or split into separate PRs — let me know what you prefer. |
完整诊断更新 — 两个 CI 阻塞均已解决 ✅1. Supply chain audit（4/26 失败 → 5/9 已通过）触发原因： 你已经在 commit - os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+ os.killpg(os.getpgid(proc.pid), getattr(signal, "SIGKILL", signal.SIGTERM))自 5/9 起审计全部 green。这个阻塞已消失，无需进一步操作。 2. Discord 测试失败（50+ errors）根因： PR 的状态： fix PR 已开： #47008 — 当前状态
结论：PR 本身干净，两个 CI 红标都不是你的代码问题。 就等 maintainer 了。你可以考虑在描述里 @ 一个活跃 maintainer（alt-glitch label 过这个 PR）让它重新进入 review 队列。 |
teknium1
reviewed
Jul 12, 2026
teknium1
left a comment
teknium1
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the DAG execution work. It needs substantial rework against the current delegation and orchestration architecture.
Problems
src/orchestration/delegate_runner.py:51passestoolsets=todelegate_task, but current main accepts no such keyword (tools/delegate_tool.py:2342-2350). Current delegation also intentionally forces children to inherit parent toolsets (tools/delegate_tool.py:2492-2494), so every DAG node currently fails before running.src/orchestration/delegate_runner.py:20mutates process-globalos.environ, whilerun_delegate_node()dispatches parallel nodes withasyncio.to_thread(src/orchestration/delegate_runner.py:81). Concurrent waves can therefore cross-contaminateprofile_envvalues.- The new coordinator, registry, and bus have no product entry point. Current main deliberately keeps durable DAG scheduling in the Kanban kernel (
hermes_cli/kanban_swarm.py:1-14) and now has an opt-in A2A plugin (6a109c84f).
Suggested changes
- Rebase the bridge concept on the current delegate_task contract, remove per-node toolset control, and add an integration-level execution test.
- Replace global environment patching with explicit child configuration, or serialize it.
- Scope the remaining feature to Kanban or the A2A plugin rather than introducing a parallel scheduler.
Automated hermes-sweeper review.
| previous: dict[str, Optional[str]] = {} | ||
| try: | ||
| for key, val in updates.items(): | ||
| previous[key] = os.environ.get(key) |
Contributor
There was a problem hiding this comment.
TaskGraph.run() executes a wave concurrently and run_delegate_node() uses asyncio.to_thread, so this process-global environment mutation can overlap another node's profile_env. Pass this configuration explicitly to child construction instead of mutating os.environ.
| raw = delegate_task( | ||
| goal=spec.goal, | ||
| context=merged_ctx or None, | ||
| toolsets=list(spec.toolsets) if spec.toolsets else None, |
Contributor
There was a problem hiding this comment.
Current delegate_task has no toolsets keyword (tools/delegate_tool.py:2342-2350), so this raises TypeError. Current main also intentionally makes delegated children inherit their parent's toolsets; remove this argument and align the graph spec with that contract.
teknium1
added
sweeper:risk-session-state
kuangmi-bit
pushed a commit
to kuangmi-bit/hermes-agent
that referenced
this pull request
Jul 18, 2026
Multi-agent orchestration with dependency-graph execution and structured pre-flight validation. Based on work by @georgex8001 (PR NousResearch#12436) with additional validation layers. TaskGraph: DAG executor with DFS cycle detection, parallel wave execution via asyncio, fail-fast propagation (dependents -> SKIPPED). validate_dag(): pre-execution validation returning structured ValidationReport: - Cycle detection (REJECT) - Orphan detection (WARNING for zero in+out degree nodes) - Depth limit (WARNING >4, REJECT >8, configurable) - Blast radius BFS (risk level: routine/elevated/critical) AgentCommunicationBus: in-process async pub/sub for A2A messaging. MultiAgentOrchestrator: coordinates via existing delegate_task. Reflex hints: record/load failure patterns for self-improvement. Stdlib only — no new dependencies. 18 tests pass. Co-authored-by: georgex8001 <georgex8001@users.noreply.github.com>
Add src/orchestration (TaskGraph, delegate bridge, A2A bus, reflex hints), wire setuptools/pytest pythonpath for the src layout, and unit tests. Rebased cleanly onto current main.
georgex8001
force-pushed
the
feat/multi-agent-orchestration-pr
branch
from
July 23, 2026 16:21
05c46b7 to
d0359e2
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds a new
orchestrationpackage undersrc/orchestration/:TaskGraph: DAG validation (cycle detection), parallel wave execution withasyncio, fail-fast propagation (dependents →SKIPPED).MultiAgentOrchestrator: orchestrates nodes via existingdelegate_task(thread offload withasyncio.to_thread), no edits todelegate_tool.py.AgentCommunicationBus: in-process async queue for Agent-to-Agent style messaging (wire to MCP transport can be layered on later).OrchestratorRegistry: hooks for custom coordinator factories.hermes_constants); state under$HERMES_HOME/orchestration/reflex_hints.jsonlviarecord_failure/load_recent_hints.Packaging
pyproject.toml:[tool.setuptools.packages.find]addswhere = [".", "src"]andorchestrationtoinclude(resolved againstorigin/mainduring cherry-pick).Tests