diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 00000000..2ed153f5
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,49 @@
+# .gitattributes -- merge strategy and diff behaviour
+#
+# Policy: documentation is merged additively (union) so two branches appending
+# to the same note never produce a conflict. Code is merged strictly, with
+# conflict markers left intact for a human to resolve.
+#
+# NOTE: merge=union is only safe for files where line order does not carry
+# meaning. It must NEVER be applied to code, config, or lock files.
+
+# --- Documentation: additive merge -----------------------------------------
+*.md      merge=union
+*.mdx     merge=union
+knowledge/**/*.md  merge=union
+
+# --- Code: strict merge, never union ---------------------------------------
+*.py      merge=text diff=python
+*.js      merge=text diff=javascript
+*.ts      merge=text diff=javascript
+*.tsx     merge=text diff=javascript
+*.yml     merge=text
+*.yaml    merge=text
+*.json    merge=text
+*.toml    merge=text
+
+# --- Generated / lock files: do not attempt a content merge ----------------
+*.lock      merge=binary -diff
+poetry.lock merge=binary -diff
+package-lock.json  merge=binary -diff
+uv.lock     merge=binary -diff
+
+# --- Never show these in diffs or language stats ---------------------------
+*.pyc       -diff linguist-generated
+__pycache__/  export-ignore
+*.min.js    -diff linguist-generated
+*.min.css   -diff linguist-generated
+
+# --- Line endings -----------------------------------------------------------
+* text=auto eol=lf
+*.ps1   text eol=crlf
+*.bat   text eol=crlf
+
+# --- Binary assets ----------------------------------------------------------
+*.png   binary
+*.jpg   binary
+*.jpeg  binary
+*.gif   binary
+*.ico   binary
+*.pdf   binary
+*.zip   binary
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
new file mode 100644
index 00000000..ab8fcd78
--- /dev/null
+++ b/.github/CODEOWNERS
@@ -0,0 +1,38 @@
+# CODEOWNERS -- who must review what
+#
+# Order matters: the LAST matching pattern wins. Put broad rules first and
+# specific overrides below them.
+#
+# Owners below are placeholders pointing at the organisation. Replace them with
+# the real team handles before relying on review routing -- an unresolvable
+# owner means GitHub falls back to "any approver" for that path.
+
+# --- Default: anything not matched below ------------------------------------
+*                       @ZyntroAI
+
+# --- Knowledge base: content requires a knowledge owner ---------------------
+/knowledge/             @ZyntroAI
+/knowledge/manifest.yml @ZyntroAI
+
+# --- Agent task skills ------------------------------------------------------
+/knowledge/agents/      @ZyntroAI
+
+# --- CI and merge protection: the highest-consequence paths -----------------
+/.github/workflows/     @ZyntroAI
+/.github/CODEOWNERS     @ZyntroAI
+/.github/merge_rules.json @ZyntroAI
+/.gitattributes         @ZyntroAI
+
+# --- Application code --------------------------------------------------------
+/app/                   @ZyntroAI
+/backend/               @ZyntroAI
+/deliverables/          @ZyntroAI
+/skills/                @ZyntroAI
+
+# --- Security-sensitive ------------------------------------------------------
+/security/              @ZyntroAI
+/.github/dependabot.yml @ZyntroAI
+
+# --- Repo-wide policy --------------------------------------------------------
+/CONTRIBUTING.md        @ZyntroAI
+/SECURITY.md            @ZyntroAI
diff --git a/.github/merge_rules.json b/.github/merge_rules.json
new file mode 100644
index 00000000..6da7a325
--- /dev/null
+++ b/.github/merge_rules.json
@@ -0,0 +1,85 @@
+{
+  "$comment": "Machine-readable merge policy for this repository. Advisory configuration: the enforcement is .github/workflows/protect-merge.yml plus the branch-protection rules on main. Keep the two in step.",
+  "version": 1,
+  "updated": "2026-09-14",
+
+  "protected_branches": ["main"],
+
+  "merge_strategy": {
+    "allowed": ["merge_commit"],
+    "default": "MERGE_COMMIT",
+    "squash_allowed": false,
+    "rebase_allowed": false,
+    "$comment": "History is preserved on main. Feature branches may squash internally, but the commit that lands on main is a merge commit."
+  },
+
+  "reviews": {
+    "required_approving_review_count": 2,
+    "dismiss_stale_reviews_on_push": true,
+    "require_review_from_codeowners": true,
+    "require_conversation_resolution": true
+  },
+
+  "status_checks": {
+    "require_branches_up_to_date": true,
+    "require_status_checks": true,
+    "required_checks": [
+      "protect-merge",
+      "CI",
+      "secret-scan"
+    ]
+  },
+
+  "branch_rules": {
+    "require_linear_history": false,
+    "allow_force_pushes": false,
+    "allow_deletions": false,
+    "require_signed_commits": false
+  },
+
+  "protected_paths": [
+    {
+      "pattern": "knowledge/**",
+      "reason": "Curated knowledge notes are content-reviewed and sha256-indexed in knowledge/manifest.yml.",
+      "rules": ["require_codeowner_review", "block_force_push"]
+    },
+    {
+      "pattern": ".github/workflows/**",
+      "reason": "CI definitions can alter what runs on main. Changes here require explicit review.",
+      "rules": ["require_codeowner_review", "block_force_push", "require_additional_approval"]
+    },
+    {
+      "pattern": ".github/CODEOWNERS",
+      "reason": "Self-protecting: the review-routing rules themselves must be reviewed.",
+      "rules": ["require_codeowner_review"]
+    },
+    {
+      "pattern": ".github/merge_rules.json",
+      "reason": "Self-protecting: the merge policy itself must be reviewed.",
+      "rules": ["require_codeowner_review"]
+    },
+    {
+      "pattern": ".gitattributes",
+      "reason": "A wrong merge strategy can silently corrupt other files' merges.",
+      "rules": ["require_codeowner_review"]
+    },
+    {
+      "pattern": "security/**",
+      "reason": "Security controls and their tests.",
+      "rules": ["require_codeowner_review"]
+    }
+  ],
+
+  "safety": {
+    "block_on_secret_detection": true,
+    "block_on_delete_of_protected_path": true,
+    "require_additive_only_for": ["knowledge/**", "docs/**"],
+    "$comment": "require_additive_only_for means files under these paths may be added or modified in a PR, but a PR that DELETES one is flagged for manual review."
+  },
+
+  "delete_protection": {
+    "enabled": true,
+    "patterns": ["knowledge/**", "docs/**", "security/**"],
+    "on_delete": "flag_for_review"
+  }
+}
diff --git a/knowledge/agents/README.md b/knowledge/agents/README.md
new file mode 100644
index 00000000..aa44d2c0
--- /dev/null
+++ b/knowledge/agents/README.md
@@ -0,0 +1,44 @@
+# Agent Task Skills
+
+Six composable skills for managing a unit of work end to end -- plan it, run it,
+grade it, unblock it, hand it over, and report on it.
+
+They are deliberately small and dependency-free: pure Python, no network, no
+state on disk. Each is a module with a dataclass vocabulary and one coordinator
+class, so they compose in any pipeline and are trivial to unit test.
+
+## Skills
+
+| Skill | Responsibility | Entry point |
+|---|---|---|
+| [task-master](task-master/SKILL.md) | Decompose a goal, order by dependency, refuse cycles | `TaskMaster.plan()` |
+| [result-orchestrator](result-orchestrator/SKILL.md) | Collect outcomes, apply quality gates, reduce to a verdict | `ResultOrchestrator.aggregate()` |
+| [milestone-tracker](milestone-tracker/SKILL.md) | Grade timeline health from slip against schedule | `MilestoneTracker.health()` |
+| [blocker-resolver](blocker-resolver/SKILL.md) | Classify obstacles and route escalations | `BlockerResolver.triage()` |
+| [handoff-coordinator](handoff-coordinator/SKILL.md) | Gate a handoff on completeness | `HandoffCoordinator.receipt()` |
+| [summary-reporter](summary-reporter/SKILL.md) | Render an executive update | `SummaryReporter.report()` |
+
+## Lifecycle
+
+```
+goal --[task-master]--> waves
+waves --[result-orchestrator]--> verdict + failure_reasons
+verdict --[milestone-tracker]--> timeline health
+health --[blocker-resolver]--> resolutions + escalations
+work --[handoff-coordinator]--> accepted | rejected
+all --[summary-reporter]--> executive update
+```
+
+## Design rules
+
+- **Deterministic.** Same input, same output -- no clocks, no randomness.
+- **Explicit failure.** A cycle, an unknown dependency, an unknown status, or an
+  incomplete handoff raises or is rejected; nothing is silently tolerated.
+- **No hidden state.** Every skill is a plain class; construct it, call it, drop it.
+- **Machine-first.** Coordinators return dicts; the reporter is the only renderer.
+
+## Running the tests
+
+```bash
+python -m pytest knowledge/agents -q
+```
diff --git a/knowledge/agents/blocker-resolver/SKILL.md b/knowledge/agents/blocker-resolver/SKILL.md
new file mode 100644
index 00000000..71a002ab
--- /dev/null
+++ b/knowledge/agents/blocker-resolver/SKILL.md
@@ -0,0 +1,49 @@
+---
+title: "Agent Skill: Blocker Resolver"
+description: "Classify obstacles, attach a resolution playbook, and route escalations."
+tags:
+  - agents/execution
+  - agents/recovery
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Blocker Resolver
+
+> Classify obstacles, attach the matching resolution playbook, and decide which ones need a human escalation before work can resume.
+
+**Module:** `knowledge.agents.blocker_resolver.main` · **Version:** 1.0.0 · **Type:** `agents/execution/recovery`
+
+## Contract
+
+`Blocker Resolver` exposes one coordinator class. Classifies an obstacle against a fixed playbook and decides whether it can be cleared locally or needs a human.
+
+## Usage
+
+```python
+from knowledge.agents.blocker_resolver.main import Blocker, BlockerResolver
+
+out = BlockerResolver().triage([
+    Blocker("b1", "missing lib", "dependency"),
+    Blocker("b2", "no workflows scope", "permission", waiting_on="repo admin"),
+])
+# {"self_resolvable": 1, "escalation_ids": ["b2"]}
+```
+
+## Design notes
+
+- Permission and unclear-requirement blockers always escalate; a machine
+  cannot grant itself a scope or invent a requirement.
+- Any blocker escalates once its retry budget is spent, so a loop can
+  never spin indefinitely on a problem it cannot solve.
+- An unrecognised category raises rather than guessing a playbook.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/blocker-resolver -q
+```
diff --git a/knowledge/agents/blocker-resolver/main.py b/knowledge/agents/blocker-resolver/main.py
new file mode 100644
index 00000000..78b64369
--- /dev/null
+++ b/knowledge/agents/blocker-resolver/main.py
@@ -0,0 +1,106 @@
+"""Blocker Resolver -- obstacle classification and recovery routing."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Sequence
+
+#: Resolution playbook keyed by blocker category.
+PLAYBOOK: Dict[str, str] = {
+    "dependency": "pin the missing dependency or vendor the interface",
+    "permission": "request the scope from the resource owner; do not retry blindly",
+    "environment": "rebuild the environment from its recorded recipe",
+    "data": "locate the authoritative source and re-derive",
+    "unclear_requirement": "return to the requester with one specific question",
+    "external_service": "wait with backoff, then report the outage with evidence",
+}
+
+#: Categories that can never be cleared without a human decision.
+ESCALATE_ALWAYS = frozenset({"permission", "unclear_requirement"})
+
+#: Attempts allowed before an otherwise-resolvable blocker escalates.
+MAX_SELF_ATTEMPTS = 2
+
+
+@dataclass
+class Blocker:
+    """One obstacle holding up work."""
+
+    id: str
+    description: str
+    category: str
+    attempts: int = 0
+    waiting_on: Optional[str] = None
+
+    def __post_init__(self) -> None:
+        if self.category not in PLAYBOOK:
+            raise ValueError(
+                f"unknown category {self.category!r}; "
+                f"expected one of {sorted(PLAYBOOK)}"
+            )
+
+
+@dataclass
+class Resolution:
+    """The recommended action for one blocker."""
+
+    blocker_id: str
+    category: str
+    action: str
+    escalate: bool
+    escalate_to: Optional[str] = None
+    owner_blocked: bool = False
+
+
+class BlockerResolver:
+    """Route blockers to a resolution or an escalation.
+
+    Args:
+        max_self_attempts: Retries before a blocker escalates anyway.
+    """
+
+    def __init__(self, max_self_attempts: int = MAX_SELF_ATTEMPTS) -> None:
+        if max_self_attempts < 0:
+            raise ValueError("max_self_attempts cannot be negative")
+        self.max_self_attempts = max_self_attempts
+
+    def resolve(self, blocker: Blocker) -> Resolution:
+        """Classify one blocker and recommend an action."""
+        must_escalate = blocker.category in ESCALATE_ALWAYS
+        exhausted = blocker.attempts >= self.max_self_attempts
+        escalate = must_escalate or exhausted
+
+        if escalate:
+            action = (
+                f"escalate to {blocker.waiting_on or 'the blocker owner'}: "
+                f"{PLAYBOOK[blocker.category]}"
+            )
+        else:
+            action = PLAYBOOK[blocker.category]
+
+        return Resolution(
+            blocker_id=blocker.id,
+            category=blocker.category,
+            action=action,
+            escalate=escalate,
+            escalate_to=blocker.waiting_on,
+            owner_blocked=blocker.waiting_on is not None,
+        )
+
+    def triage(self, blockers: Sequence[Blocker]) -> Dict[str, object]:
+        """Resolve a whole set and split what you can fix from what you cannot."""
+        resolutions = [self.resolve(b) for b in blockers]
+        needs_human = [r for r in resolutions if r.escalate]
+        return {
+            "total": len(blockers),
+            "self_resolvable": len(resolutions) - len(needs_human),
+            "needs_escalation": len(needs_human),
+            "escalation_ids": [r.blocker_id for r in needs_human],
+            "resolutions": {
+                r.blocker_id: {
+                    "category": r.category,
+                    "action": r.action,
+                    "escalate": r.escalate,
+                }
+                for r in resolutions
+            },
+        }
diff --git a/knowledge/agents/blocker-resolver/tests/test_blocker_resolver.py b/knowledge/agents/blocker-resolver/tests/test_blocker_resolver.py
new file mode 100644
index 00000000..8ee4e40d
--- /dev/null
+++ b/knowledge/agents/blocker-resolver/tests/test_blocker_resolver.py
@@ -0,0 +1,72 @@
+"""Tests for the blocker-resolver skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_blocker_resolver"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+BlockerResolver = _skill.BlockerResolver
+Blocker = _skill.Blocker
+
+
+def test_dependency_is_self_resolvable():
+    res = BlockerResolver().resolve(Blocker("b1", "missing lib", "dependency"))
+    assert res.escalate is False
+    assert "pin the missing dependency" in res.action
+
+
+def test_permission_always_escalates():
+    res = BlockerResolver().resolve(
+        Blocker("b2", "no workflows scope", "permission", waiting_on="repo admin")
+    )
+    assert res.escalate is True
+    assert res.escalate_to == "repo admin"
+    assert res.owner_blocked is True
+
+
+def test_exhausted_attempts_escalate():
+    res = BlockerResolver(max_self_attempts=2).resolve(
+        Blocker("b3", "flaky env", "environment", attempts=2)
+    )
+    assert res.escalate is True
+
+
+def test_triage_splits_the_set():
+    blockers = [
+        Blocker("b1", "missing lib", "dependency"),
+        Blocker("b2", "no scope", "permission"),
+        Blocker("b3", "bad data", "data"),
+    ]
+    out = BlockerResolver().triage(blockers)
+    assert out["self_resolvable"] == 2
+    assert out["needs_escalation"] == 1
+    assert out["escalation_ids"] == ["b2"]
+
+
+def test_unknown_category_is_rejected():
+    with pytest.raises(ValueError):
+        BlockerResolver().resolve(Blocker("b4", "weird", "cosmic_rays"))
+
+
+def test_empty_triage():
+    out = BlockerResolver().triage([])
+    assert out["total"] == 0
+    assert out["needs_escalation"] == 0
diff --git a/knowledge/agents/handoff-coordinator/SKILL.md b/knowledge/agents/handoff-coordinator/SKILL.md
new file mode 100644
index 00000000..1dcb9aff
--- /dev/null
+++ b/knowledge/agents/handoff-coordinator/SKILL.md
@@ -0,0 +1,50 @@
+---
+title: "Agent Skill: Handoff Coordinator"
+description: "Validate a handoff payload before work crosses an owner boundary."
+tags:
+  - agents/coordination
+  - agents/handoff
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Handoff Coordinator
+
+> Validate a handoff payload before work passes to another team. Rejects an incomplete handoff rather than letting it across the gap.
+
+**Module:** `knowledge.agents.handoff_coordinator.main` &middot; **Version:** 1.0.0 &middot; **Type:** `agents/coordination/handoff`
+
+## Contract
+
+`Handoff Coordinator` exposes one coordinator class. Validates that a handoff carries everything the receiver needs before it crosses an owner boundary.
+
+## Usage
+
+```python
+from knowledge.agents.handoff_coordinator.main import Handoff, HandoffCoordinator
+
+receipt = HandoffCoordinator().receipt(Handoff(
+    work_id="W-1", from_owner="backend", to_owner="frontend",
+    summary="API contract finalized", artifacts=["openapi.json"],
+    acceptance_criteria=["returns 200 on /health"],
+    next_action="wire the client to /health",
+))
+# {"status": "accepted", "receiver_next_step": "wire the client to /health"}
+```
+
+## Design notes
+
+- Four fields are mandatory: summary, artifacts, acceptance_criteria,
+  next_action. A handoff missing any of them is rejected, not trimmed.
+- Handing off to yourself is rejected -- it is a state change, not a handoff.
+- `require_open_questions=True` blocks a handoff carrying unresolved questions.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/handoff-coordinator -q
+```
diff --git a/knowledge/agents/handoff-coordinator/main.py b/knowledge/agents/handoff-coordinator/main.py
new file mode 100644
index 00000000..33a7517f
--- /dev/null
+++ b/knowledge/agents/handoff-coordinator/main.py
@@ -0,0 +1,91 @@
+"""Handoff Coordinator -- validate work passing between owners."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Sequence
+
+#: Fields every handoff must carry for the receiver to act.
+REQUIRED_FIELDS = (
+    "summary",
+    "artifacts",
+    "acceptance_criteria",
+    "next_action",
+)
+
+
+@dataclass
+class Handoff:
+    """A unit of work passing from one owner to another."""
+
+    work_id: str
+    from_owner: str
+    to_owner: str
+    summary: str = ""
+    artifacts: List[str] = field(default_factory=list)
+    acceptance_criteria: List[str] = field(default_factory=list)
+    next_action: str = ""
+    open_questions: List[str] = field(default_factory=list)
+
+    def field_value(self, name: str) -> object:
+        return getattr(self, name)
+
+
+class HandoffCoordinator:
+    """Gate handoffs on completeness.
+
+    Args:
+        require_open_questions: When true, unresolved questions block the handoff.
+    """
+
+    def __init__(self, require_open_questions: bool = False) -> None:
+        self.require_open_questions = require_open_questions
+
+    def validate(self, handoff: Handoff) -> List[str]:
+        """Return a list of problems; empty means the handoff is ready."""
+        problems: List[str] = []
+        for name in REQUIRED_FIELDS:
+            value = handoff.field_value(name)
+            if isinstance(value, list):
+                if not value:
+                    problems.append(f"{name} is empty")
+            elif not str(value).strip():
+                problems.append(f"{name} is empty")
+        if not handoff.to_owner.strip():
+            problems.append("to_owner is empty")
+        if handoff.from_owner == handoff.to_owner:
+            problems.append("from_owner and to_owner are the same")
+        if self.require_open_questions and handoff.open_questions:
+            problems.append(
+                f"{len(handoff.open_questions)} unresolved question(s) must be closed"
+            )
+        return problems
+
+    def accept(self, handoff: Handoff) -> bool:
+        """True when the handoff may proceed."""
+        return not self.validate(handoff)
+
+    def receipt(self, handoff: Handoff) -> Dict[str, object]:
+        """Render an accept/reject receipt the receiver can act on."""
+        problems = self.validate(handoff)
+        return {
+            "work_id": handoff.work_id,
+            "from": handoff.from_owner,
+            "to": handoff.to_owner,
+            "status": "accepted" if not problems else "rejected",
+            "problems": problems,
+            "artifact_count": len(handoff.artifacts),
+            "criteria_count": len(handoff.acceptance_criteria),
+            "receiver_next_step": handoff.next_action if not problems else None,
+        }
+
+    def batch(self, handoffs: Sequence[Handoff]) -> Dict[str, object]:
+        """Summarize a set of handoffs."""
+        receipts = [self.receipt(h) for h in handoffs]
+        rejected = [r["work_id"] for r in receipts if r["status"] == "rejected"]
+        return {
+            "total": len(handoffs),
+            "accepted": len(receipts) - len(rejected),
+            "rejected": len(rejected),
+            "rejected_ids": sorted(rejected),
+            "receipts": receipts,
+        }
diff --git a/knowledge/agents/handoff-coordinator/tests/test_handoff_coordinator.py b/knowledge/agents/handoff-coordinator/tests/test_handoff_coordinator.py
new file mode 100644
index 00000000..22aa376a
--- /dev/null
+++ b/knowledge/agents/handoff-coordinator/tests/test_handoff_coordinator.py
@@ -0,0 +1,78 @@
+"""Tests for the handoff-coordinator skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_handoff_coordinator"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+HandoffCoordinator = _skill.HandoffCoordinator
+Handoff = _skill.Handoff
+
+
+def _complete(**kw):
+    base = dict(
+        work_id="W-1",
+        from_owner="backend",
+        to_owner="frontend",
+        summary="API contract finalized",
+        artifacts=["openapi.json"],
+        acceptance_criteria=["returns 200 on /health"],
+        next_action="wire the client to /health",
+    )
+    base.update(kw)
+    return Handoff(**base)
+
+
+def test_complete_handoff_is_accepted():
+    assert HandoffCoordinator().accept(_complete()) is True
+
+
+def test_missing_summary_is_rejected():
+    problems = HandoffCoordinator().validate(_complete(summary="   "))
+    assert any("summary" in p for p in problems)
+
+
+def test_missing_artifacts_is_rejected():
+    problems = HandoffCoordinator().validate(_complete(artifacts=[]))
+    assert any("artifacts" in p for p in problems)
+
+
+def test_same_owner_is_rejected():
+    problems = HandoffCoordinator().validate(_complete(to_owner="backend"))
+    assert any("same" in p for p in problems)
+
+
+def test_open_questions_block_only_when_required():
+    h = _complete(open_questions=["which auth scheme?"])
+    assert HandoffCoordinator().accept(h) is True
+    assert HandoffCoordinator(require_open_questions=True).accept(h) is False
+
+
+def test_receipt_points_at_next_step():
+    receipt = HandoffCoordinator().receipt(_complete())
+    assert receipt["status"] == "accepted"
+    assert receipt["receiver_next_step"] == "wire the client to /health"
+
+
+def test_batch_counts_rejections():
+    out = HandoffCoordinator().batch([_complete(), _complete(work_id="W-2", summary="")])
+    assert out["accepted"] == 1
+    assert out["rejected_ids"] == ["W-2"]
diff --git a/knowledge/agents/manifest.yml b/knowledge/agents/manifest.yml
new file mode 100644
index 00000000..98e05a7b
--- /dev/null
+++ b/knowledge/agents/manifest.yml
@@ -0,0 +1,60 @@
+# knowledge/agents/manifest.yml -- machine-readable index of agent task skills
+# Each skill is a self-contained module: SKILL.md (contract) + main.py (logic) + tests/.
+
+version: 1
+kind: agent-skills
+conventions:
+  layout: "<skill>/SKILL.md + <skill>/main.py + <skill>/tests/test_<skill>.py"
+  entrypoint: "<skill>.main"
+  tests: pytest
+  python: ">=3.12"
+
+skills:
+- id: task-master
+  name: Task Master
+  path: task-master
+  module: knowledge.agents.task_master.main
+  type: agents/planning/decomposition
+  responsibility: "Decompose a goal into subtasks, order by dependency, refuse cycles."
+  inputs: [goal, subtasks]
+  outputs: [waves, critical_path_hours, plan]
+- id: result-orchestrator
+  name: Result Orchestrator
+  path: result-orchestrator
+  module: knowledge.agents.result_orchestrator.main
+  type: agents/execution/orchestration
+  responsibility: "Collect outcomes, apply quality gates, reduce to one verdict."
+  inputs: [subtask_results]
+  outputs: [verdict, pass_ratio, failure_reasons]
+- id: milestone-tracker
+  name: Milestone Tracker
+  path: milestone-tracker
+  module: knowledge.agents.milestone_tracker.main
+  type: agents/planning/tracking
+  responsibility: "Grade phase timeline health from slip against schedule."
+  inputs: [milestones]
+  outputs: [overall, grade_counts, total_slip_days]
+- id: blocker-resolver
+  name: Blocker Resolver
+  path: blocker-resolver
+  module: knowledge.agents.blocker_resolver.main
+  type: agents/execution/recovery
+  responsibility: "Classify obstacles, attach a playbook, route escalations."
+  inputs: [blockers]
+  outputs: [resolutions, escalation_ids]
+- id: handoff-coordinator
+  name: Handoff Coordinator
+  path: handoff-coordinator
+  module: knowledge.agents.handoff_coordinator.main
+  type: agents/coordination/handoff
+  responsibility: "Validate a handoff payload before work crosses an owner boundary."
+  inputs: [handoff]
+  outputs: [status, problems, receiver_next_step]
+- id: summary-reporter
+  name: Summary Reporter
+  path: summary-reporter
+  module: knowledge.agents.summary_reporter.main
+  type: agents/communication/reporting
+  responsibility: "Render an executive update that leads with required decisions."
+  inputs: [tasks]
+  outputs: [headline, markdown_report]
diff --git a/knowledge/agents/milestone-tracker/SKILL.md b/knowledge/agents/milestone-tracker/SKILL.md
new file mode 100644
index 00000000..0eaff19c
--- /dev/null
+++ b/knowledge/agents/milestone-tracker/SKILL.md
@@ -0,0 +1,47 @@
+---
+title: "Agent Skill: Milestone Tracker"
+description: "Grade phase timeline health from slip against schedule and burn."
+tags:
+  - agents/planning
+  - agents/tracking
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Milestone Tracker
+
+> Track phases against a plan and grade timeline health from slip against schedule and elapsed-time burn.
+
+**Module:** `knowledge.agents.milestone_tracker.main` &middot; **Version:** 1.0.0 &middot; **Type:** `agents/planning/tracking`
+
+## Contract
+
+`Milestone Tracker` exposes one coordinator class. Grades each phase of a plan from slip against its schedule, then rolls the grades up into one health verdict.
+
+## Usage
+
+```python
+from knowledge.agents.milestone_tracker.main import Milestone, MilestoneTracker
+
+health = MilestoneTracker().health([
+    Milestone("m1", "Schema", planned_days=10, actual_days=10, complete=True),
+    Milestone("m2", "API", planned_days=10, actual_days=15),
+])
+# {"overall": "delayed", "percent_complete": 50.0, "total_slip_days": 5.0}
+```
+
+## Design notes
+
+- Grades are `complete | delayed | at_risk | on_track`; the overall
+  rollup takes the worst of them, so one slipping phase is never averaged away.
+- Thresholds are configurable but validated -- at-risk must sit below delayed.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/milestone-tracker -q
+```
diff --git a/knowledge/agents/milestone-tracker/main.py b/knowledge/agents/milestone-tracker/main.py
new file mode 100644
index 00000000..edb73ef7
--- /dev/null
+++ b/knowledge/agents/milestone-tracker/main.py
@@ -0,0 +1,92 @@
+"""Milestone Tracker -- phase progress and timeline health."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Sequence
+
+
+@dataclass
+class Milestone:
+    """One phase of a plan."""
+
+    id: str
+    name: str
+    planned_days: float
+    actual_days: float = 0.0
+    complete: bool = False
+
+    @property
+    def slip_days(self) -> float:
+        """Days over (positive) or under (negative) the plan."""
+        return round(self.actual_days - self.planned_days, 2)
+
+    @property
+    def progress(self) -> float:
+        """Fraction of the planned duration consumed, capped at 1.0."""
+        if self.planned_days <= 0:
+            return 1.0 if self.complete else 0.0
+        return round(min(self.actual_days / self.planned_days, 1.0), 4)
+
+
+class MilestoneTracker:
+    """Grade timeline health across phases.
+
+    Args:
+        at_risk_slip_ratio: Slip fraction that escalates a phase to at-risk.
+        delayed_slip_ratio: Slip fraction that escalates a phase to delayed.
+    """
+
+    def __init__(
+        self, at_risk_slip_ratio: float = 0.15, delayed_slip_ratio: float = 0.4
+    ) -> None:
+        if at_risk_slip_ratio >= delayed_slip_ratio:
+            raise ValueError("at_risk threshold must be below the delayed threshold")
+        self.at_risk_slip_ratio = at_risk_slip_ratio
+        self.delayed_slip_ratio = delayed_slip_ratio
+
+    def grade(self, milestone: Milestone) -> str:
+        """Grade one milestone: complete | delayed | at_risk | on_track."""
+        if milestone.complete:
+            return "complete"
+        if milestone.planned_days <= 0:
+            return "at_risk"
+        ratio = milestone.slip_days / milestone.planned_days
+        if ratio >= self.delayed_slip_ratio:
+            return "delayed"
+        if ratio >= self.at_risk_slip_ratio:
+            return "at_risk"
+        return "on_track"
+
+    def health(self, milestones: Sequence[Milestone]) -> Dict[str, object]:
+        """Aggregate phase grades into a rollup."""
+        grades = {m.id: self.grade(m) for m in milestones}
+        counts: Dict[str, int] = {}
+        for grade in grades.values():
+            counts[grade] = counts.get(grade, 0) + 1
+
+        done = sum(1 for m in milestones if m.complete)
+        total = len(milestones)
+        overall = "complete" if total and done == total else "in_progress"
+        if not total:
+            overall = "empty"
+        elif counts.get("delayed"):
+            overall = "delayed"
+        elif counts.get("at_risk"):
+            overall = "at_risk"
+
+        return {
+            "overall": overall,
+            "milestones": total,
+            "complete": done,
+            "percent_complete": round((done / total * 100) if total else 0.0, 1),
+            "grade_counts": counts,
+            "grades": grades,
+            "total_slip_days": round(sum(m.slip_days for m in milestones), 2),
+        }
+
+    def at_risk_ids(self, milestones: Sequence[Milestone]) -> List[str]:
+        """Ids graded at_risk or delayed, worst slip first."""
+        risky = [
+            m for m in milestones if self.grade(m) in ("at_risk", "delayed")
+        ]
+        return [m.id for m in sorted(risky, key=lambda m: -m.slip_days)]
diff --git a/knowledge/agents/milestone-tracker/tests/test_milestone_tracker.py b/knowledge/agents/milestone-tracker/tests/test_milestone_tracker.py
new file mode 100644
index 00000000..3508978a
--- /dev/null
+++ b/knowledge/agents/milestone-tracker/tests/test_milestone_tracker.py
@@ -0,0 +1,68 @@
+"""Tests for the milestone-tracker skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_milestone_tracker"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+MilestoneTracker = _skill.MilestoneTracker
+Milestone = _skill.Milestone
+
+
+def test_complete_beats_slip():
+    m = Milestone("a", "Phase A", planned_days=10, actual_days=20, complete=True)
+    assert MilestoneTracker().grade(m) == "complete"
+
+
+def test_on_track_inside_threshold():
+    m = Milestone("a", "Phase A", planned_days=10, actual_days=11)
+    assert MilestoneTracker().grade(m) == "on_track"
+
+
+def test_at_risk_and_delayed_thresholds():
+    tracker = MilestoneTracker()
+    assert tracker.grade(Milestone("a", "A", 10, 12)) == "at_risk"
+    assert tracker.grade(Milestone("b", "B", 10, 15)) == "delayed"
+
+
+def test_health_rollup_flags_delay():
+    ms = [
+        Milestone("a", "A", 10, 10, complete=True),
+        Milestone("b", "B", 10, 15),
+    ]
+    out = MilestoneTracker().health(ms)
+    assert out["overall"] == "delayed"
+    assert out["percent_complete"] == 50.0
+    assert out["total_slip_days"] == 5.0
+
+
+def test_at_risk_ids_sorted_by_slip():
+    ms = [Milestone("a", "A", 10, 12), Milestone("b", "B", 10, 16)]
+    assert MilestoneTracker().at_risk_ids(ms) == ["b", "a"]
+
+
+def test_empty_is_explicit():
+    assert MilestoneTracker().health([])["overall"] == "empty"
+
+
+def test_bad_thresholds_rejected():
+    with pytest.raises(ValueError):
+        MilestoneTracker(at_risk_slip_ratio=0.5, delayed_slip_ratio=0.2)
diff --git a/knowledge/agents/result-orchestrator/SKILL.md b/knowledge/agents/result-orchestrator/SKILL.md
new file mode 100644
index 00000000..b5b77932
--- /dev/null
+++ b/knowledge/agents/result-orchestrator/SKILL.md
@@ -0,0 +1,49 @@
+---
+title: "Agent Skill: Result Orchestrator"
+description: "Collect subtask outcomes, apply quality gates, and reduce them to one verdict."
+tags:
+  - agents/execution
+  - agents/orchestration
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Result Orchestrator
+
+> Collect subtask outcomes, apply quality gates, and reduce them to a single verdict. A subtask only counts as passing when it succeeded AND cleared every gate.
+
+**Module:** `knowledge.agents.result_orchestrator.main` &middot; **Version:** 1.0.0 &middot; **Type:** `agents/execution/orchestration`
+
+## Contract
+
+`Result Orchestrator` exposes one coordinator class. Reduces many subtask outcomes into a single verdict. A subtask counts as passing only when it succeeded **and** every quality gate passed.
+
+## Usage
+
+```python
+from knowledge.agents.result_orchestrator.main import (
+    GateResult, ResultOrchestrator, SubtaskResult)
+
+out = ResultOrchestrator(min_pass_ratio=0.8).aggregate([
+    SubtaskResult("api", True, [GateResult("tests", True)]),
+    SubtaskResult("ui", True, [GateResult("lint", False, "3 errors")]),
+])
+# {"verdict": "failed", "failure_reasons": {"ui": "quality gate failed: lint"}}
+```
+
+## Design notes
+
+- A failed gate is named in `failure_reasons`; a raw error is preserved
+  ahead of gate detail, because it is the more actionable signal.
+- An empty input is `empty`, never `passed` -- silence is not success.
+- `blocked_by()` returns the ids that must be fixed before the job can pass.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/result-orchestrator -q
+```
diff --git a/knowledge/agents/result-orchestrator/main.py b/knowledge/agents/result-orchestrator/main.py
new file mode 100644
index 00000000..446d4548
--- /dev/null
+++ b/knowledge/agents/result-orchestrator/main.py
@@ -0,0 +1,81 @@
+"""Result Orchestrator -- outcome aggregation and quality gates."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Sequence
+
+
+@dataclass
+class GateResult:
+    """One quality gate applied to a subtask result."""
+
+    name: str
+    passed: bool
+    detail: str = ""
+
+
+@dataclass
+class SubtaskResult:
+    """The outcome of one subtask."""
+
+    id: str
+    ok: bool
+    gates: List[GateResult] = field(default_factory=list)
+    error: Optional[str] = None
+
+    @property
+    def passed(self) -> bool:
+        """True only when the task succeeded and every gate passed."""
+        return self.ok and all(g.passed for g in self.gates)
+
+    def failed_gates(self) -> List[str]:
+        return [g.name for g in self.gates if not g.passed]
+
+
+class ResultOrchestrator:
+    """Reduce many subtask results into one verdict.
+
+    Args:
+        min_pass_ratio: Fraction of subtasks that must pass to report success.
+    """
+
+    def __init__(self, min_pass_ratio: float = 1.0) -> None:
+        if not 0.0 <= min_pass_ratio <= 1.0:
+            raise ValueError("min_pass_ratio must be between 0 and 1")
+        self.min_pass_ratio = min_pass_ratio
+
+    def aggregate(self, results: Sequence[SubtaskResult]) -> Dict[str, object]:
+        """Reduce results to a verdict dict with per-task reasons."""
+        total = len(results)
+        passed = [r for r in results if r.passed]
+        failed = [r for r in results if not r.passed]
+
+        reasons: Dict[str, str] = {}
+        for r in failed:
+            if not r.ok:
+                reasons[r.id] = r.error or "subtask reported failure"
+            else:
+                reasons[r.id] = "quality gate failed: " + ", ".join(r.failed_gates())
+
+        ratio = (len(passed) / total) if total else 1.0
+        if total == 0:
+            verdict = "empty"
+        elif not failed:
+            verdict = "passed"
+        elif ratio >= self.min_pass_ratio:
+            verdict = "passed_with_warnings"
+        else:
+            verdict = "failed"
+
+        return {
+            "verdict": verdict,
+            "total": total,
+            "passed": len(passed),
+            "failed": len(failed),
+            "pass_ratio": round(ratio, 4),
+            "failure_reasons": reasons,
+        }
+
+    def blocked_by(self, results: Sequence[SubtaskResult]) -> List[str]:
+        """Ids of subtasks that must be fixed before the whole job can pass."""
+        return sorted(r.id for r in results if not r.passed)
diff --git a/knowledge/agents/result-orchestrator/tests/test_result_orchestrator.py b/knowledge/agents/result-orchestrator/tests/test_result_orchestrator.py
new file mode 100644
index 00000000..e39a22fa
--- /dev/null
+++ b/knowledge/agents/result-orchestrator/tests/test_result_orchestrator.py
@@ -0,0 +1,75 @@
+"""Tests for the result-orchestrator skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_result_orchestrator"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+ResultOrchestrator = _skill.ResultOrchestrator
+SubtaskResult = _skill.SubtaskResult
+GateResult = _skill.GateResult
+
+
+def test_all_pass_is_clean():
+    results = [
+        SubtaskResult("a", True, [GateResult("tests", True)]),
+        SubtaskResult("b", True, [GateResult("lint", True)]),
+    ]
+    out = ResultOrchestrator().aggregate(results)
+    assert out["verdict"] == "passed"
+    assert out["failure_reasons"] == {}
+
+
+def test_failed_gate_marks_task_failed():
+    results = [SubtaskResult("a", True, [GateResult("tests", False, "3 red")])]
+    out = ResultOrchestrator().aggregate(results)
+    assert out["verdict"] == "failed"
+    assert "quality gate failed: tests" in out["failure_reasons"]["a"]
+
+
+def test_error_is_surfaced_over_gates():
+    results = [SubtaskResult("a", False, error="timeout")]
+    out = ResultOrchestrator().aggregate(results)
+    assert out["failure_reasons"]["a"] == "timeout"
+
+
+def test_threshold_allows_partial_pass():
+    results = [
+        SubtaskResult("a", True),
+        SubtaskResult("b", False, error="boom"),
+    ]
+    out = ResultOrchestrator(min_pass_ratio=0.5).aggregate(results)
+    assert out["verdict"] == "passed_with_warnings"
+    assert out["pass_ratio"] == 0.5
+
+
+def test_empty_input_is_not_a_pass():
+    assert ResultOrchestrator().aggregate([])["verdict"] == "empty"
+
+
+def test_blocked_by_lists_only_failures():
+    results = [SubtaskResult("a", True), SubtaskResult("b", False, error="x")]
+    assert ResultOrchestrator().blocked_by(results) == ["b"]
+
+
+def test_bad_ratio_is_rejected():
+    with pytest.raises(ValueError):
+        ResultOrchestrator(min_pass_ratio=1.5)
diff --git a/knowledge/agents/summary-reporter/SKILL.md b/knowledge/agents/summary-reporter/SKILL.md
new file mode 100644
index 00000000..c76e5ca3
--- /dev/null
+++ b/knowledge/agents/summary-reporter/SKILL.md
@@ -0,0 +1,50 @@
+---
+title: "Agent Skill: Summary Reporter"
+description: "Render an executive update from task state, leading with required decisions."
+tags:
+  - agents/communication
+  - agents/reporting
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Summary Reporter
+
+> Render an executive update from task state -- lead with what needs a decision, then status, then detail. Never buries a blocker.
+
+**Module:** `knowledge.agents.summary_reporter.main` &middot; **Version:** 1.0.0 &middot; **Type:** `agents/communication/reporting`
+
+## Contract
+
+`Summary Reporter` exposes one coordinator class. Renders a structured task list into an executive update that leads with what needs a human decision.
+
+## Usage
+
+```python
+from knowledge.agents.summary_reporter.main import SummaryReporter, TaskState
+
+md = SummaryReporter().report("Weekly update", [
+    TaskState("T-1", "Ship auth", "done"),
+    TaskState("T-2", "Wire billing", "blocked", blocker="waiting on keys"),
+])
+# Headline: "1 item(s) blocked; 0 decision(s) needed from you."
+```
+
+## Design notes
+
+- The order is fixed: headline, decisions, blockers, status, detail. A
+  reader who stops after the first line still knows what to do.
+- The reporter is the only renderer in the suite; everything else returns
+  machine-readable dicts, so this is the one place formatting lives.
+- Detail rows are truncated with an explicit remainder count, never a
+  silent cut.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/summary-reporter -q
+```
diff --git a/knowledge/agents/summary-reporter/main.py b/knowledge/agents/summary-reporter/main.py
new file mode 100644
index 00000000..3156ad20
--- /dev/null
+++ b/knowledge/agents/summary-reporter/main.py
@@ -0,0 +1,101 @@
+"""Summary Reporter -- executive updates from structured task state."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Sequence
+
+
+@dataclass
+class TaskState:
+    """Flattened state for one work item."""
+
+    id: str
+    title: str
+    status: str
+    blocker: str = ""
+    needs_decision: str = ""
+
+    def __post_init__(self) -> None:
+        if self.status not in ("done", "in_progress", "blocked", "not_started"):
+            raise ValueError(f"{self.id}: unknown status {self.status!r}")
+
+
+class SummaryReporter:
+    """Build an executive summary that leads with what needs a human.
+
+    Args:
+        max_detail_items: Rows of raw detail to include.
+    """
+
+    def __init__(self, max_detail_items: int = 10) -> None:
+        self.max_detail_items = max_detail_items
+
+    def counts(self, tasks: Sequence[TaskState]) -> Dict[str, int]:
+        out = {"done": 0, "in_progress": 0, "blocked": 0, "not_started": 0}
+        for t in tasks:
+            out[t.status] += 1
+        return out
+
+    def decisions_needed(self, tasks: Sequence[TaskState]) -> List[Dict[str, str]]:
+        """Every item waiting on a human, most blocked first."""
+        waiting = [
+            {"id": t.id, "title": t.title, "ask": t.needs_decision}
+            for t in tasks
+            if t.needs_decision.strip()
+        ]
+        return waiting
+
+    def headline(self, tasks: Sequence[TaskState]) -> str:
+        """One line an executive can read on its own."""
+        if not tasks:
+            return "No work in flight."
+        c = self.counts(tasks)
+        decisions = len(self.decisions_needed(tasks))
+        if c["blocked"]:
+            return (
+                f"{c['blocked']} item(s) blocked; "
+                f"{decisions} decision(s) needed from you."
+            )
+        if decisions:
+            return f"On track; {decisions} decision(s) needed from you."
+        if c["done"] == len(tasks):
+            return "All items complete."
+        return f"On track; {c['in_progress']} item(s) in progress."
+
+    def report(self, title: str, tasks: Sequence[TaskState]) -> str:
+        """Render the full markdown update."""
+        lines = [f"# {title}", "", self.headline(tasks), ""]
+
+        decisions = self.decisions_needed(tasks)
+        if decisions:
+            lines += ["## Needs your decision", ""]
+            for d in decisions:
+                lines.append(f"- **{d['id']}** -- {d['ask']} ({d['title']})")
+            lines.append("")
+
+        blocked = [t for t in tasks if t.status == "blocked"]
+        if blocked:
+            lines += ["## Blocked", ""]
+            for t in blocked:
+                reason = t.blocker or "reason not recorded"
+                lines.append(f"- **{t.id}** -- {t.title}: {reason}")
+            lines.append("")
+
+        c = self.counts(tasks)
+        lines += [
+            "## Status",
+            "",
+            f"- Done: {c['done']}",
+            f"- In progress: {c['in_progress']}",
+            f"- Blocked: {c['blocked']}",
+            f"- Not started: {c['not_started']}",
+            "",
+        ]
+
+        detail = tasks[: self.max_detail_items]
+        lines += ["## Detail", ""]
+        for t in detail:
+            lines.append(f"- {t.id} -- {t.title} [{t.status}]")
+        if len(tasks) > len(detail):
+            lines.append(f"- ...and {len(tasks) - len(detail)} more")
+        return "\n".join(lines)
diff --git a/knowledge/agents/summary-reporter/tests/test_summary_reporter.py b/knowledge/agents/summary-reporter/tests/test_summary_reporter.py
new file mode 100644
index 00000000..859aae22
--- /dev/null
+++ b/knowledge/agents/summary-reporter/tests/test_summary_reporter.py
@@ -0,0 +1,75 @@
+"""Tests for the summary-reporter skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_summary_reporter"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+SummaryReporter = _skill.SummaryReporter
+TaskState = _skill.TaskState
+
+
+def _tasks():
+    return [
+        TaskState("T-1", "Ship auth", "done"),
+        TaskState("T-2", "Wire billing", "blocked", blocker="waiting on Stripe keys"),
+        TaskState("T-3", "Pick database", "in_progress", needs_decision="Postgres or MySQL?"),
+    ]
+
+
+def test_headline_leads_with_blockers_and_decisions():
+    headline = SummaryReporter().headline(_tasks())
+    assert "1 item(s) blocked" in headline
+    assert "1 decision(s) needed" in headline
+
+
+def test_report_puts_decisions_before_status():
+    report = SummaryReporter().report("Weekly update", _tasks())
+    assert report.index("## Needs your decision") < report.index("## Status")
+
+
+def test_blocked_section_names_the_reason():
+    report = SummaryReporter().report("Weekly update", _tasks())
+    assert "waiting on Stripe keys" in report
+
+
+def test_counts_are_exact():
+    assert SummaryReporter().counts(_tasks()) == {
+        "done": 1,
+        "in_progress": 1,
+        "blocked": 1,
+        "not_started": 0,
+    }
+
+
+def test_detail_is_truncated():
+    tasks = [TaskState(f"T-{i}", "x", "done") for i in range(15)]
+    report = SummaryReporter(max_detail_items=10).report("Big", tasks)
+    assert "...and 5 more" in report
+
+
+def test_empty_report_is_honest():
+    assert SummaryReporter().headline([]) == "No work in flight."
+
+
+def test_unknown_status_rejected():
+    with pytest.raises(ValueError):
+        TaskState("T-9", "x", "vibing")
diff --git a/knowledge/agents/task-master/SKILL.md b/knowledge/agents/task-master/SKILL.md
new file mode 100644
index 00000000..912661f2
--- /dev/null
+++ b/knowledge/agents/task-master/SKILL.md
@@ -0,0 +1,49 @@
+---
+title: "Agent Skill: Task Master"
+description: "Decompose a goal into atomic subtasks, order by dependency, and refuse cycles."
+tags:
+  - agents/planning
+  - agents/decomposition
+  - agents/skills
+doc_kind: "skill"
+status: "active"
+owner: "Platform Engineering"
+last_reviewed: "2026-09-13"
+review_frequency: "Annual"
+---
+
+# Task Master
+
+> Break a goal into atomic subtasks, resolve dependencies, and emit a deterministic execution order. Refuses to schedule a plan containing a dependency cycle.
+
+**Module:** `knowledge.agents.task_master.main` &middot; **Version:** 1.0.0 &middot; **Type:** `agents/planning/decomposition`
+
+## Contract
+
+`Task Master` exposes one coordinator class. Decomposes a goal into atomic subtasks and returns a wave-ordered execution plan. A wave contains only subtasks whose dependencies are already satisfied, so every wave can run fully in parallel.
+
+## Usage
+
+```python
+from knowledge.agents.task_master.main import Subtask, TaskMaster
+
+plan = TaskMaster().plan("Ship the feature", [
+    Subtask("design", "Design schema", estimate_hours=3),
+    Subtask("api", "Build API", depends_on=["design"], estimate_hours=5),
+    Subtask("ship", "Ship", depends_on=["api"], estimate_hours=1),
+])
+# {"waves": [["design"], ["api"], ["ship"]], "critical_path_hours": 9.0, ...}
+```
+
+## Design notes
+
+- A circular dependency raises `DependencyCycleError` -- the plan is
+  never emitted in a state that cannot be executed.
+- `critical_path_hours` is the longest dependency chain, which is the
+  floor on wall-clock time regardless of available parallelism.
+
+## Tests
+
+```bash
+python -m pytest knowledge/agents/task-master -q
+```
diff --git a/knowledge/agents/task-master/main.py b/knowledge/agents/task-master/main.py
new file mode 100644
index 00000000..9c32001e
--- /dev/null
+++ b/knowledge/agents/task-master/main.py
@@ -0,0 +1,107 @@
+"""Task Master -- goal decomposition and dependency ordering."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Sequence
+
+
+class DependencyCycleError(ValueError):
+    """Raised when subtasks declare a circular dependency."""
+
+
+@dataclass
+class Subtask:
+    """One atomic unit of work."""
+
+    id: str
+    title: str
+    depends_on: List[str] = field(default_factory=list)
+    estimate_hours: float = 1.0
+
+    def __post_init__(self) -> None:
+        if not self.id:
+            raise ValueError("subtask id is required")
+        if self.estimate_hours <= 0:
+            raise ValueError(f"{self.id}: estimate_hours must be positive")
+
+
+class TaskMaster:
+    """Decompose a goal and order the parts.
+
+    Args:
+        max_parallel: Upper bound on subtasks reported as simultaneously ready.
+    """
+
+    def __init__(self, max_parallel: int = 4) -> None:
+        if max_parallel < 1:
+            raise ValueError("max_parallel must be >= 1")
+        self.max_parallel = max_parallel
+
+    def validate(self, subtasks: Sequence[Subtask]) -> None:
+        """Check ids are unique and every dependency resolves to a known subtask."""
+        seen: Dict[str, bool] = {}
+        for task in subtasks:
+            if task.id in seen:
+                raise ValueError(f"duplicate subtask id: {task.id}")
+            seen[task.id] = True
+        for task in subtasks:
+            for dep in task.depends_on:
+                if dep not in seen:
+                    raise ValueError(f"{task.id} depends on unknown subtask {dep}")
+
+    def execution_order(self, subtasks: Sequence[Subtask]) -> List[List[str]]:
+        """Group subtasks into waves; every task in a wave is dependency-free.
+
+        Returns a list of waves, each a sorted list of subtask ids. Raises
+        ``DependencyCycleError`` if no valid ordering exists.
+        """
+        self.validate(subtasks)
+        remaining = {t.id: set(t.depends_on) for t in subtasks}
+        waves: List[List[str]] = []
+        placed: set[str] = set()
+
+        while remaining:
+            ready = sorted(
+                tid for tid, deps in remaining.items() if deps <= placed
+            )
+            if not ready:
+                blocked = sorted(remaining)
+                raise DependencyCycleError(
+                    "circular dependency among: " + ", ".join(blocked)
+                )
+            waves.append(ready)
+            placed.update(ready)
+            for tid in ready:
+                del remaining[tid]
+        return waves
+
+    def critical_path_hours(self, subtasks: Sequence[Subtask]) -> float:
+        """Longest dependency chain by estimate -- the floor on wall-clock time."""
+        self.validate(subtasks)
+        by_id = {t.id: t for t in subtasks}
+        memo: Dict[str, float] = {}
+
+        def depth(tid: str) -> float:
+            if tid in memo:
+                return memo[tid]
+            task = by_id[tid]
+            best = 0.0
+            for dep in task.depends_on:
+                best = max(best, depth(dep))
+            memo[tid] = best + task.estimate_hours
+            return memo[tid]
+
+        return round(max((depth(t.id) for t in subtasks), default=0.0), 2)
+
+    def plan(self, goal: str, subtasks: Sequence[Subtask]) -> Dict[str, object]:
+        """Render the full plan as a machine-readable dict."""
+        waves = self.execution_order(subtasks)
+        return {
+            "goal": goal,
+            "subtask_count": len(subtasks),
+            "waves": waves,
+            "wave_count": len(waves),
+            "parallel_width": max((len(w) for w in waves), default=0),
+            "critical_path_hours": self.critical_path_hours(subtasks),
+            "max_parallel": self.max_parallel,
+        }
diff --git a/knowledge/agents/task-master/tests/test_task_master.py b/knowledge/agents/task-master/tests/test_task_master.py
new file mode 100644
index 00000000..491ee7ad
--- /dev/null
+++ b/knowledge/agents/task-master/tests/test_task_master.py
@@ -0,0 +1,71 @@
+"""Tests for the task-master skill."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _load():
+    """Load the skill's main.py by path -- kebab-case dirs are not packages."""
+    main_py = Path(__file__).resolve().parents[1] / "main.py"
+    mod_name = "agent_skill_task_master"
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    spec = importlib.util.spec_from_file_location(mod_name, main_py)
+    assert spec and spec.loader, f"cannot load {main_py}"
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[mod_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_skill = _load()
+TaskMaster = _skill.TaskMaster
+Subtask = _skill.Subtask
+DependencyCycleError = _skill.DependencyCycleError
+
+
+def _tasks():
+    return [
+        Subtask("design", "Design the schema", estimate_hours=3),
+        Subtask("api", "Build the API", depends_on=["design"], estimate_hours=5),
+        Subtask("docs", "Write the docs", depends_on=["design"], estimate_hours=2),
+        Subtask("ship", "Ship it", depends_on=["api", "docs"], estimate_hours=1),
+    ]
+
+
+def test_waves_respect_dependencies():
+    waves = TaskMaster().execution_order(_tasks())
+    assert waves == [["design"], ["api", "docs"], ["ship"]]
+
+
+def test_critical_path_is_longest_chain():
+    assert TaskMaster().critical_path_hours(_tasks()) == 9.0
+
+
+def test_cycle_is_rejected():
+    cyclic = [
+        Subtask("a", "A", depends_on=["b"]),
+        Subtask("b", "B", depends_on=["a"]),
+    ]
+    with pytest.raises(DependencyCycleError):
+        TaskMaster().execution_order(cyclic)
+
+
+def test_unknown_dependency_is_rejected():
+    with pytest.raises(ValueError):
+        TaskMaster().execution_order([Subtask("a", "A", depends_on=["ghost"])])
+
+
+def test_duplicate_id_is_rejected():
+    with pytest.raises(ValueError):
+        TaskMaster().execution_order([Subtask("a", "A"), Subtask("a", "A again")])
+
+
+def test_plan_reports_parallel_width():
+    plan = TaskMaster().plan("Ship the feature", _tasks())
+    assert plan["wave_count"] == 3
+    assert plan["parallel_width"] == 2