1#!/usr/bin/env python3
2"""Compositional router over catalog.json (plan phase P3).
3
4Turns a recommended product list plus DETECTED context (platform, cloud) into an
5ordered onboarding plan by resolving category-level `requires` against the
6capability graph and topologically ordering the bound dependency edges. `kind`
7is only the deterministic tie-breaker for otherwise-independent nodes. No
8intent-to-skill table exists anywhere: intents live only in the recommender;
9here we compose.
10
11CLI:
12  python3 resolve.py --products "APM,Infrastructure Monitoring" --platform kubernetes --cloud aws
13"""
14
15import argparse

resolve.py on GitHub

import difflib

import json

import os

import re

import sys

import uuid

HERE = os.path.dirname(os.path.abspath(__file__)) # …/dd-orchestrator/scripts

SKILL = os.path.dirname(HERE) # …/dd-orchestrator

CATALOG = os.path.join(SKILL, "catalog.json")

KIND_RANK = {"foundation": 0, "cloud-connect": 1, "platform-install": 2,

"product-enable": 3, "verify-troubleshoot": 4, "lifecycle": 5}

# recommender product name / alias -> catalog product token

PRODUCT_TOKENS = {

"apm": "apm", "application performance monitoring": "apm",

"llm observability": "llm-obs", "llmo": "llm-obs", "llm obs": "llm-obs",

"rum": "rum", "real user monitoring": "rum", "session replay": "rum",

"error tracking": "error-tracking", "product analytics": "product-analytics",

"infrastructure monitoring": "infra", "infra": "infra", "container monitoring": "infra",

"log management": "logs", "logs": "logs",

"database monitoring": "dbm", "dbm": "dbm",

"cloud cost management": "cost", "ccm": "cost", "cloud cost": "cost",

"cloud security": "cloud-security", "csm": "cloud-security", "cspm": "cloud-security", "cws": "cloud-security",

"cloud siem": "cloud-siem",

"app and api protection": "aap", "aap": "aap", "asm": "aap",

"test optimization": "test-optimization", "ci visibility": "test-optimization",

"code coverage": "code-coverage",

"synthetic monitoring": "synthetics", "synthetics": "synthetics",

"continuous profiler": "profiler", "profiler": "profiler",

"network device monitoring": "ndm", "sensitive data scanner": "sds",

"slo": "slo", "monitors": "monitors", "opentelemetry": "otel", "otel": "otel",

"studio": "studio", "storage": "storage", "agent observability": "agent-observability",

"mobile rum": "rum-mobile", "rum mobile": "rum-mobile", "mobile monitoring": "rum-mobile",

"datadog app": "apps", "datadog apps": "apps", "dashboards app": "apps",

}

# detected platform sub-variants -> the catalog platform token the Agent skills use.

# 'fargate' is deliberately NOT aliased: EKS-Fargate is kubernetes but ECS-Fargate is ecs, so

# collapsing it would pick the wrong installer (ROB-7); leave it to a dead-end/choice instead.

PLATFORM_ALIASES = {

"k8s": "kubernetes", "eks": "kubernetes", "gke": "kubernetes", "aks": "kubernetes",

"openshift": "kubernetes", "oke": "kubernetes", "tkg": "kubernetes", "rancher": "kubernetes",

"autopilot": "kubernetes",

# Generic hosts/VMs do not imply Linux, and EC2/GCE do not reveal the guest

# operating system. GCR is a registry, not the Cloud Run compute platform.

"cloudrun": "cloud-run",

# Serverless/PaaS deploy targets have no Agent installer; collapse the common ones to a

# single 'serverless' class so Agent-based products dead-end honestly (with a pointer)

# instead of offering an install choice that cannot match the context. (spec req #12)

"vercel": "serverless", "netlify": "serverless",

}

CLOUD_ALIASES = {

"amazon": "aws", "amazon web services": "aws", "google": "gcp", "google cloud": "gcp",

"microsoft azure": "azure", "az": "azure",

}

def normalize_platform(p):

# ROB-2: lowercase the TOKEN itself (not just the lookup key), so canonical-but-

# capitalized inputs like "Kubernetes"/"Docker" resolve instead of dead-ending.

p2 = (p or "").strip().lower()

return PLATFORM_ALIASES.get(p2, p2)

def normalize_cloud(c):

c2 = (c or "").strip().lower()

return CLOUD_ALIASES.get(c2, c2)

# Serverless/PaaS platforms Datadog cannot instrument with an Agent installer. A product

# that needs a platform-install dead-ends here (with a pointer), never as a choice. (req #12)

SERVERLESS_PLATFORMS = {"serverless"}

def _serverless_pointer(platform):

"""Pointer appended to a platform dead-end when the platform is serverless/PaaS

(no Agent installer exists). Empty string for every other platform."""

if platform in SERVERLESS_PLATFORMS:

return (" — serverless/PaaS has no Agent installer; use the agentless integration "

"(e.g. the Vercel/Netlify <-> Datadog integration) or an agentless SDK, "

"see docs.datadoghq.com")

return ""

def _cloud_suffix(cloud):

"""Shared ' for cloud X' clause appended to a platform dead-end message, or ''."""

return f" for cloud '{_san(cloud)}'" if cloud and cloud != "none" else ""

def _san(s, n=80):

"""ROB-8: sanitize free-text before echoing it into dead-end/choice strings."""

s = "".join(ch if (ch >= " " and ch != "\x7f") else " " for ch in str(s))

s = " ".join(s.split())

return (s[:n] + "…") if len(s) > n else s

def load_catalog(path=CATALOG):

with open(path) as f:

return json.load(f)

def canonical_key(node, by_key):

"""Walk `duplicate_of` links to the canonical (non-duplicate) root key.

`by_key` maps every node's `key` to the node itself. Guards against cycles;

malformed catalogs that would cycle are rejected by check_catalog before

this ever runs on real data.

"""

key, seen = node["key"], set()

while by_key.get(key, {}).get("duplicate_of"):

if key in seen:

break

seen.add(key)

key = by_key[key]["duplicate_of"]

return key

def display_name(token):

"""First human alias in PRODUCT_TOKENS that maps to `token` (else the token itself)."""

return next((k for k, v in PRODUCT_TOKENS.items() if v == token), token)

def normalize_product(name, catalog_tokens=None):

"""Map a product name/alias to its catalog token, or None if unrecognized.

Accepts the recommender names/aliases in PRODUCT_TOKENS and, when a set of

catalog product tokens is supplied, every such token as an alias for itself

(so an operator can pass the exact token printed in catalog.json). Case- and

whitespace-insensitive.

"""

key = str(name).strip().lower()

if key in PRODUCT_TOKENS:

return PRODUCT_TOKENS[key]

if catalog_tokens and key in catalog_tokens:

return key

return None

# Orchestrator machinery, not user-facing products. Excluded from intent detection so

# "recommend what I need" (the canonical vague case) is never read as a named product.

_NON_PRODUCT_TOKENS = frozenset({"account", "orchestrate", "recommend"})

def detect_products(intent, catalog_tokens=None):

"""Scan a free-text intent for EXPLICITLY NAMED products; return their catalog tokens

ordered by first mention. Empty result => no product named => the caller falls back to

the recommender (spec req "Ability to get a shortcut and skip some steps").

Matches product NAMES, not the concepts they cover (word-boundary, longest-phrase-first):

"Agent Observability" (a name) matches; "track LLM model calls" (a description) does not.

The vocabulary is exactly what --products accepts (PRODUCT_TOKENS aliases + the catalog's

own tokens), minus orchestrator machinery — so it never drifts from the resolver.

"""

if not intent:

return []

vocab = (set(PRODUCT_TOKENS) | set(catalog_tokens or [])) - _NON_PRODUCT_TOKENS

# Longest phrase first so "mobile rum" wins over "rum"; the matched span is then consumed

# so a contained shorter alias cannot double-match. Fully ordered key => deterministic.

terms = sorted(vocab, key=lambda t: (-t.count(" "), -len(t), t))

original = intent.lower()

haystack = original

first_pos = {}

for term in terms:

pattern = r"\b" + r"\s+".join(re.escape(w) for w in term.split()) + r"\b"

if not re.search(pattern, haystack):

continue

token = normalize_product(term, catalog_tokens)

if token:

here = re.search(pattern, original)

pos = here.start() if here else 0

first_pos[token] = min(pos, first_pos.get(token, pos))

haystack = re.sub(pattern, " ", haystack) # consume so a shorter alias can't re-match

return [t for t, _ in sorted(first_pos.items(), key=lambda kv: (kv[1], kv[0]))]

class Router:

def __init__(self, catalog, enabled_only=True):

# A canonical node identifies a capability; it is not necessarily the

# implementation available at runtime. Choose one implementation per duplicate

# group, preferring the canonical when it is enabled and otherwise falling back

# to a stable enabled duplicate. This prevents a disabled capability identity

# from hiding a genuinely equivalent enabled implementation.

nodes = [n for n in catalog["nodes"] if n["kind"] != "internal"]

by_key = {n["key"]: n for n in nodes}

# Every catalog product token is a first-class --products alias for itself, so an

# operator can pass the exact token printed in catalog.json (spec req #11). Derived

# from the catalog so it stays in sync as the catalog evolves.

self.product_tokens = frozenset(

n["product"].strip().lower() for n in nodes if n.get("product")

)

groups = {}

for n in nodes:

groups.setdefault(canonical_key(n, by_key), []).append(n)

self.live = []

self.alias_ids_by_key = {}

self.by_id = {}

for root, implementations in groups.items():

canonical = by_key.get(root)

if enabled_only:

available = [n for n in implementations if n.get("enabled", True)]

if not available:

continue

chosen = canonical if canonical in available else min(available, key=lambda n: (n["id"], n["key"]))

else:

chosen = canonical or min(implementations, key=lambda n: (n["id"], n["key"]))

self.live.append(chosen)

aliases = {n["id"] for n in implementations}

self.alias_ids_by_key[chosen["key"]] = aliases

for alias in aliases:

self.by_id[alias] = chosen

self.account = next((n for n in self.live if n["product"] == "account"), None)

self.install = {}

self.cloud = {}

self.enable = {}

self.verify = {}

self.triggered = {}

self.installer_products = set()

for n in self.live:

if n["kind"] == "platform-install":

for platform in n["platform"]:

self.install.setdefault(platform, []).append(n)

self.installer_products.update(n.get("delivers", []))

if n["kind"] == "cloud-connect" and n["action"] == "connect":

for cloud in n["cloud"]:

self.cloud.setdefault(cloud, []).append(n)

if n["kind"] == "product-enable":

self.enable.setdefault(n["product"], []).append(n)

if n["kind"] == "verify-troubleshoot" and "verify" in n["action"]:

if n["product"]:

self.verify.setdefault(n["product"], []).append(n)

for product in n.get("trigger_products") or []:

self.triggered.setdefault(product, []).append(n)

for index in (self.install, self.cloud, self.enable, self.verify, self.triggered):

for values in index.values():

values.sort(key=lambda n: (n["id"], n["key"]))

@staticmethod

def _facet_rank(node):

"""Rank a matching node by constraint specificity, not list length."""

constrained = int(bool(node["platform"])) + int(bool(node["cloud"]))

return (constrained, -len(node["platform"]), -len(node["cloud"]))

@staticmethod

def _first(candidates):

return candidates[0] if candidates else None

def _bind_soft(self, cat, ctx, fixed_cloud=None):

"""Bind a suggests edge; returns None (silently) if it cannot bind."""

if cat == "cloud-connect":

c = fixed_cloud or ctx.get("cloud")

return self._first(self.cloud.get(c, [])) if c and c != "none" else None

if cat == "platform-install":

p = ctx.get("platform")

candidates = [n for n in self.install.get(p, []) if self._fits(n, ctx)]

return self._first(candidates) if p and p != "none" else None

if cat == "foundation":

return self.account

return None

def _match(self, cands, ctx):

"""Pick the candidate whose platform/cloud facets fit the detected context."""

best, best_score = None, None

for n in cands:

if n["platform"] and ctx.get("platform") not in n["platform"]:

continue

if n["cloud"] and ctx.get("cloud") not in n["cloud"]:

continue

score = self._facet_rank(n)

if best_score is None or score > best_score:

best, best_score = n, score

return best

@staticmethod

def _fits(node, ctx):

return ((not node["platform"] or ctx.get("platform") in node["platform"]) and

(not node["cloud"] or ctx.get("cloud") in node["cloud"]))

def _nearest_product(self, name):

"""Nearest known product for an unrecognized input: (display, token), or None.

Searches recommender names/aliases plus catalog tokens and returns the closest

only when it is a genuine near-miss (difflib cutoff 0.6): a typo like

"llmobs" -> "llm-obs" gets a "did you mean" hint, while garbage with no real

similarity returns None so the dead-end carries no fabricated suggestion.

"""

vocab = list(PRODUCT_TOKENS) + sorted(self.product_tokens)

match = difflib.get_close_matches(str(name).strip().lower(), vocab, n=1, cutoff=0.6)

if not match:

return None

token = normalize_product(match[0], self.product_tokens)

return display_name(token), token

def resolve(self, products, ctx, selections=None):

"""Resolve products for context.

``selections`` maps a product token/name (or a returned choice ``need``) to

a skill id. It lets callers resolve a genuine capability choice without

relying on catalog order, e.g. ``{"agent-observability": "trace-rca"}``.

For convenience, the same mapping may be supplied as ``ctx["selections"]``.

"""

raw_selections = dict(ctx.get("selections") or {})

raw_selections.update(selections or {})

selected = {}

for name, skill_id in raw_selections.items():

normalized = normalize_product(str(name), self.product_tokens) or str(name).strip().lower()

selected[normalized] = str(skill_id).strip()

selected[str(name).strip().lower()] = str(skill_id).strip()

ctx = {**ctx,

"platform": normalize_platform(ctx.get("platform")), # ROB-2: "Kubernetes"/"EKS" -> kubernetes

"cloud": normalize_cloud(ctx.get("cloud"))}

include, dead_ends, choice_options, caveats = {}, [], {}, []

dependencies = set() # (dependent key, prerequisite key)

def add_choice(need, options):

options = set(options)

if need in choice_options:

# One detected context must satisfy every dependent capability.

choice_options[need].intersection_update(options)

else:

choice_options[need] = options

def requested_selection(*names):

for name in names:

if name and name.lower() in selected:

return selected[name.lower()]

return None

def accepts_id(node, skill_id):

return skill_id == node["id"] or skill_id in self.alias_ids_by_key.get(node["key"], set())

def choose_implementation(candidates, need, *selection_names):

"""Choose among equally applicable, non-duplicate capabilities."""

candidates = sorted(candidates, key=lambda n: (n["id"], n["key"]))

if not candidates:

return None

if len(candidates) == 1:

return candidates[0]

selection = requested_selection(*selection_names, need)

if selection:

matches = [n for n in candidates if accepts_id(n, selection)]

if len(matches) == 1:

return matches[0]

dead_ends.append(f"selection '{_san(selection)}' is not available for {_san(need)}")

add_choice(need, (n["id"] for n in candidates))

return None

def product_candidate(tok, label):

cands = self.enable.get(tok, [])

if not cands:

dead_ends.append(f"{_san(label)} — no setup skill yet (not automated)")

return None

matches = [n for n in cands if fits(n)]

if not matches:

plats = sorted({p for n in cands for p in n["platform"]})

clouds = sorted({c for n in cands for c in n["cloud"]})

if plats and (not ctx.get("platform") or ctx.get("platform") == "none"):

add_choice("platform", plats)

elif clouds and (not ctx.get("cloud") or ctx.get("cloud") == "none"):

add_choice("cloud", clouds)

else:

where = f"platform='{_san(ctx.get('platform'))}'" if plats else f"cloud='{_san(ctx.get('cloud'))}'"

covered = f" (covered: {', '.join(plats or clouds)})" if (plats or clouds) else ""

pointer = _serverless_pointer(ctx.get("platform")) if plats else ""

dead_ends.append(f"{_san(label)} — not automated for {where}{covered}{pointer}")

return None

selection = requested_selection(tok, f"{tok} capability")

if selection:

selected_matches = [n for n in matches if accepts_id(n, selection)]

if len(selected_matches) == 1:

return selected_matches[0]

dead_ends.append(f"selection '{_san(selection)}' is not available for {_san(tok)} capability")

add_choice(f"{tok} capability", (n["id"] for n in matches))

return None

best = max(self._facet_rank(n) for n in matches)

top = [n for n in matches if self._facet_rank(n) == best]

return choose_implementation(top, f"{tok} capability", tok)

def add(n):

if n:

include[n["key"]] = n

if n.get("caveat"):

caveats.append(f"{n['id']}: {n['caveat']}")

def fits(n):

return self._fits(n, ctx)

def bind(edge):

# Returns (node|None, ok). ok=False means a HARD prerequisite could not be met,

# so the dependent must NOT be planned (bind-then-add — CORR-1). Choice points

# are recorded as a side effect and also return ok=False.

if "skill" in edge:

t = self.by_id.get(edge["skill"]); return (t, t is not None)

if "product" in edge: # CORR-4: product -> product

t = product_candidate(edge["product"], f"prerequisite '{edge['product']}'")

return (t, t is not None)

cat = edge.get("category")

if cat == "foundation":

return (self.account, self.account is not None)

if cat == "platform-install":

p = ctx.get("platform")

if not p or p == "none":

add_choice("platform", self.install); return (None, False)

candidates = [n for n in self.install.get(p, []) if fits(n)]

generic = [n for n in candidates if n["action"] in ("install", "install-agent")]

n = choose_implementation(

generic or candidates,

f"platform-install:{p} capability",

f"platform-install:{p}",

)

if not n:

if not candidates:

cloud = ctx.get("cloud")

suffix = _cloud_suffix(cloud)

dead_ends.append(

f"Agent install for platform '{_san(p)}'{suffix} is not automated"

+ _serverless_pointer(p))

return (n, n is not None)

if cat == "cloud-connect":

c = edge.get("cloud") or ctx.get("cloud")

if not c or c == "none":

add_choice("cloud", self.cloud); return (None, False)

n = choose_implementation(

self.cloud.get(c, []),

f"cloud-connect:{c} capability",

f"cloud-connect:{c}",

)

if not n:

if c not in self.cloud:

dead_ends.append(f"cloud integration for '{_san(c)}' is not automated")

return (n, n is not None)

return (None, True) # unknown edge: not a hard blocker

def plan_node(n, visiting, visited):

# (ok, staged, edges): stage n and all satisfiable hard prerequisites.

# Keep visiting separate from visited so a real dependency cycle is a

# hard failure rather than being silently treated as a shared node.

if n["key"] in visiting:

start = visiting.index(n["key"])

cycle = visiting[start:] + [n["key"]]

dead_ends.append("dependency cycle: " + " -> ".join(cycle))

return (False, {}, set())

if n["key"] in visited:

return (True, {}, set())

visiting.append(n["key"])

staged, edges, ok = {n["key"]: n}, set(), True

for r in n["requires"]:

t, edge_ok = bind(r)

if not edge_ok:

ok = False; continue

if t:

edges.add((n["key"], t["key"]))

sub_ok, sub, sub_edges = plan_node(t, visiting, visited)

staged.update(sub)

edges.update(sub_edges)

ok = ok and sub_ok

# A product pulled in as a hard prerequisite must complete its

# onboarding verification before the dependent runs, just as it

# would when requested directly.

if "product" in r:

for verifier in self.verify.get(r["product"], []):

if not fits(verifier):

continue

edges.add((n["key"], verifier["key"]))

verify_ok, verify_nodes, verify_edges = plan_node(

verifier, visiting, visited)

staged.update(verify_nodes)

edges.update(verify_edges)

ok = ok and verify_ok

visiting.pop()

visited.add(n["key"])

return (ok, staged, edges)

def stage(node, label):

dead_end_count = len(dead_ends)

choices_before = {need: set(options) for need, options in choice_options.items()}

ok, staged, edges = plan_node(node, [], set())

if ok:

for v in staged.values():

add(v)

dependencies.update(edges)

return True

# Preserve a precise bind error or an actionable choice without adding

# a second, generic dead-end that tells the user nothing new.

choices_changed = choices_before != choice_options

if len(dead_ends) == dead_end_count and not choices_changed:

dead_ends.append(

f"{_san(label)} — not set up: a required prerequisite is unavailable for this context")

return False

# Process products in a canonical order so the plan is INDEPENDENT of input order:

# installer-delivered products (infra/logs) go LAST, so they reuse an already-staged

# specific installer (e.g. apm-agent-install-*) instead of adding a redundant generic

# agent install. Non-installer products keep a stable token order. (combinatorial invariant)

def _proc_order(nm):

t = normalize_product(nm, self.product_tokens)

return (1 if t in self.installer_products else 0, t or "", str(nm))

for name in sorted(products, key=_proc_order):

tok = normalize_product(name, self.product_tokens)

if tok is None:

guess = self._nearest_product(name)

hint = f' — did you mean "{guess[0]}" ({guess[1]})?' if guess else ""

dead_ends.append(f"unrecognized product '{_san(name)}'{hint}")

continue

if tok in self.installer_products: # delivered by an installer; no separate enable skill

p = ctx.get("platform")

if not p or p == "none":

add_choice("platform", self.install); continue

candidates = []

for candidate in self.install.get(p, []):

delivered = candidate.get("delivers", [])

if tok in delivered and fits(candidate):

candidates.append(candidate)

already_staged = [n for n in candidates if n["key"] in include]

generic = [n for n in candidates if n["action"] in ("install", "install-agent")]

node = choose_implementation(

already_staged or generic or candidates,

f"platform-install:{p} capability",

f"platform-install:{p}",

)

if not node:

if not candidates:

cloud = ctx.get("cloud")

suffix = _cloud_suffix(cloud)

if not self.install.get(p):

dead_ends.append(

f"{_san(name)} — Agent install for platform '{_san(p)}' is not automated"

+ _serverless_pointer(p))

elif not any(fits(candidate) for candidate in self.install.get(p, [])):

dead_ends.append(

f"{_san(name)} — Agent install for platform '{_san(p)}'{suffix} "

"is not automated")

else:

dead_ends.append(

f"{_san(name)} — Agent install for platform '{_san(p)}'{suffix} "

"does not deliver this product")

continue

# Delivery is catalog data, not resolver policy. A platform installer

# without an explicit contract cannot be assumed to enable any product.

delivered_products = node.get("delivers", [])

if tok not in delivered_products:

dead_ends.append(f"{_san(name)} — Agent install for platform '{_san(p)}' does not deliver this product")

continue

if stage(node, name):

# Some products have additive, context-specific setup paths. For

# example, cloud log forwarding belongs in a Logs plan only when

# that cloud was actually detected. Its own hard prerequisites are

# staged normally; it is never suggested for unrelated products.

for triggered in self.triggered.get(tok, []):

if fits(triggered):

stage(triggered, f"{name} ({triggered['id']})")

continue

m = product_candidate(tok, name)

if not m:

continue

if stage(m, name): # bind-then-add: only attach verify if the product itself resolved

for v in self.verify.get(tok, []):

if not fits(v):

continue

add(v)

dependencies.add((v["key"], m["key"]))

def stable_key(key):

n = include[key]

return (KIND_RANK[n["kind"]], n.get("product") or "", n["id"], n["key"])

# Linearize into a deterministic, sequential plan with dependency-chain

# LOCALITY: after a skill completes, its newly-unblocked direct dependents

# are preferred over unrelated nodes that were already ready, so a chain

# stays contiguous instead of interleaving with a sibling branch. Ties fall

# back to the stable key (kind, product, id, key). This is a depth-first walk

# of the hard-edge DAG; it stays independent of product input order because

# every choice is made by stable_key, not arrival order. `kind` still only

# orders nodes not already constrained by a hard edge (e.g. AAP->APM).

pending = {key: set() for key in include}

dependents = {key: [] for key in include}

for dependent, prerequisite in dependencies:

if dependent in pending and prerequisite in pending:

pending[dependent].add(prerequisite)

dependents[prerequisite].append(dependent)

plan = []

emitted = set()

# LIFO frontier of ready keys. Push newly-ready keys in REVERSE stable order

# so the stable-min is popped first; the LIFO discipline drains the most

# recently unblocked chain before returning to older ready siblings.

frontier = sorted((k for k, deps in pending.items() if not deps),

key=stable_key, reverse=True)

while frontier:

key = frontier.pop()

if key in emitted:

continue

plan.append(include[key])

emitted.add(key)

newly = []

for dep in dependents[key]:

pending[dep].discard(key)

if not pending[dep] and dep not in emitted:

newly.append(dep)

frontier.extend(sorted(newly, key=stable_key, reverse=True))

if len(emitted) != len(include):

cycle = sorted((k for k in include if k not in emitted), key=stable_key)

dead_ends.append("dependency cycle among planned skills: " + ", ".join(cycle))

# dedupe by id, preserve order

seen_ids, ordered, ordered_keys = set(), [], []

for n in plan:

if n["id"] in seen_ids:

continue

seen_ids.add(n["id"])

ordered_keys.append(n["key"])

ordered.append({"id": n["id"], "kind": n["kind"], "product": n["product"],

"platform": n["platform"], "cloud": n["cloud"],

"url": n["source"]["url"], "source": n["source"]})

# A compact, human-readable view of the same plan as a DAG. Nodes stay in

# topological order, and `requires` contains only direct prerequisites.

position = {n["key"]: i for i, n in enumerate(plan)}

requires_by_key = {n["key"]: set() for n in plan}

for dependent, prerequisite in dependencies:

if dependent in requires_by_key and prerequisite in position:

requires_by_key[dependent].add(prerequisite)

dag = [

{

"skill": include[key]["id"],

"requires": [

include[required]["id"]

for required in sorted(requires_by_key[key], key=position.get)

],

}

for key in ordered_keys

]

# optional enrichments (suggests) — bound to context, never a dead-end/choice

plan_ids = {x["id"] for x in ordered}

suggested = {}

for n in include.values():

for s in n.get("suggests", []):

t = self.by_id.get(s["skill"]) if "skill" in s else self._bind_soft(s.get("category"), ctx, s.get("cloud"))

if t and t["id"] not in plan_ids and t["key"] not in suggested:

suggested[t["key"]] = {"id": t["id"], "kind": t["kind"], "product": t["product"],

"url": t["source"]["url"], "source": t["source"],

"suggested_by": n["id"]}

return {"plan": ordered,

"dag": dag,

"suggested": list(suggested.values()),

"caveats": list(dict.fromkeys(caveats)),

"dead_ends": list(dict.fromkeys(dead_ends)),

"choices": [{"need": need, "options": sorted(options)}

for need, options in choice_options.items()]}

def _print(res):

# Render the plan as an aligned table so a debugger user sees each skill AND its

# full source URL in a separate column before anything is dispatched (spec req

# "Skill source URLs in output"). Rows keep the ` N. <id>` shape parsers rely on.

plan = res["plan"]

print("PLAN (in order) — skills to execute, with source URLs:")

if plan:

rows = [(f"{i}.", p["id"],

p["kind"] + (f"/{p['product']}" if p["product"] else ""),

p["url"]) for i, p in enumerate(plan, 1)]

nw = max([len(r[0]) for r in rows] + [1])

iw = max([len(r[1]) for r in rows] + [len("SKILL")])

kw = max([len(r[2]) for r in rows] + [len("KIND")])

print(f" {'#'.ljust(nw)} {'SKILL'.ljust(iw)} {'KIND'.ljust(kw)} URL")

for num, sid, kind, url in rows:

print(f" {num.ljust(nw)} {sid.ljust(iw)} {kind.ljust(kw)} {url}")

else:

print(" (no skills to execute)")

if res.get("suggested"):

print("SUGGESTED (optional enrichment for the detected context):")

for s in res["suggested"]:

print(f" + {s['id']} [{s['kind']}] (suggested by {s['suggested_by']}) {s['url']}")

if res.get("caveats"):

print("CAVEATS (scope limits of a routed skill):")

for c in res["caveats"]:

print(f" ! {c}")

if res["dead_ends"]:

print("DEAD-ENDS (recommended, not automated — demand signal):")

for d in res["dead_ends"]:

print(f" - {d}")

if res["choices"]:

print("CHOICE POINTS (ask the developer):")

for c in res["choices"]:

print(f" - pick a {c['need']}: {', '.join(c['options'])}")

def _print_trace(session_id, args, res):

"""Emit a stable, machine-readable TRACE block the SKILL.md runbook pastes verbatim

into the run's trace file (T2.1). The judge grades that trace, so having the deterministic

planner author it — instead of the model re-narrating the plan — removes a whole class

of transcription error. Deterministic for fixed inputs; SESSION_ID is the only

run-seeded field (pin it with DD_ORCH_SESSION_ID). CONFIRMED and DISPATCHED are

placeholders the agent fills after the confirm gate and after each dispatch.

STOP_REASON tells the reader whether the plan may proceed:

none — plan is non-empty and no choices remain; go to the confirm gate.

awaiting_choice — a CHOICE_POINT is unresolved (e.g. missing platform); resolve it

and re-run first, even if a partial plan already exists.

no_enabled_capability — no plan and no choice; only dead-ends (nothing to automate).

no_plan — nothing to do (empty product list / all filtered out).

"""

plan = res["plan"]

if res["choices"]: # an unresolved choice blocks dispatch, even with a partial plan

stop = "awaiting_choice"

elif plan:

stop = "none"

elif res["dead_ends"]:

stop = "no_enabled_capability"

else:

stop = "no_plan"

plat = normalize_platform(args.platform) or "none"

cl = normalize_cloud(args.cloud) or "none"

print("=== DD-ORCH TRACE v1 ===")

print(f"SESSION_ID: {session_id}")

print(f"CONTEXT: platform={plat} cloud={cl}")

print(f"STOP_REASON: {stop}")

if plan:

print("PLAN:")

for i, p in enumerate(plan, 1):

print(f" {i} {p['id']} {p['kind']} {p['product'] or '-'} {p['url']}")

else:

print("PLAN: (empty)")

if res["dead_ends"]:

print("DEAD_ENDS:")

for d in res["dead_ends"]:

print(f" - {d}")

else:

print("DEAD_ENDS: (none)")

if res["choices"]:

print("CHOICE_POINTS:")

for c in res["choices"]:

print(f" - {c['need']}: {', '.join(c['options'])}")

else:

print("CHOICE_POINTS: (none)")

if res.get("suggested"):

print("SUGGESTED:") # optional enrichment, e.g. a cloud connector

for s in res["suggested"]:

print(f" + {s['id']} {s['kind']} (by {s['suggested_by']}) {s['url']}")

else:

print("SUGGESTED: (none)")

print("CONFIRMED: pending")

print("DISPATCHED: (fill one skill_id per line after each dispatch)")

print("=== END DD-ORCH TRACE ===")

def _print_dag(res):

"""Render the plan as an ASCII DAG (debug mode): indentation encodes dependency

depth so locality-ordered chains read as staircases, and `<-` lists each skill's

direct prerequisites so multi-parent edges stay explicit."""

dag = res["dag"]

if not dag:

print("DAG: (empty — no skills to execute)")

return

order = [n["skill"] for n in dag]

reqs = {n["skill"]: n["requires"] for n in dag}

depth = {}

for skill in order: # plan order guarantees prerequisites precede dependents

rs = reqs[skill]

depth[skill] = 1 + max(depth[r] for r in rs) if rs else 0

print("DAG (skills to execute - indent = dependency depth, `<-` = direct prerequisites):")

for skill in order:

edge = f" <- {', '.join(reqs[skill])}" if reqs[skill] else ""

print(f"{' ' * (depth[skill] + 1)}{skill}{edge}")

def _emit_telemetry(session_id, args, products, router, res):

"""Emit resolve.py's three process-guaranteed events, best-effort (v1).

This is the RELIABLE CORE of the orchestrator's telemetry — the SKILL.md runbook

drives the per-dispatch skill_step events, which a markdown runbook cannot make

process-grade. It never affects the resolve result or the exit path: emit.py is

best-effort and the whole body is guarded, so any failure is swallowed.

"""

try:

import emit

base = {

"invocation_mode": "orchestrated",

"entry_skill_id": "dd-orchestrator",

"agent_name": getattr(args, "agent_name", None) or "unknown",

"intent_mode": getattr(args, "intent_mode", None), # explicit (shortcut) | recommended

"target_platform": normalize_platform(args.platform),

"target_cloud": normalize_cloud(args.cloud),

"org_id": os.environ.get("DD_ORG_ID") or None, # auth'd org public_id; None -> omitted

}

# Persist the run envelope once so the SKILL.md runbook's later emit.py processes

# re-attach it to every started/finished/skill_run:finished event (F4).

emit.write_session_state(session_id, base, shape={

"dead_end_count": len(res["dead_ends"]),

"planned_skill_count": len(res["plan"]),

"choice_count": len(res["choices"]),

})

emit.emit("skill_run", "started", session_id, dict(base), critical=True)

tokens = sorted({t for t in (normalize_product(p, router.product_tokens)

for p in products) if t})

dependency_count = {node["skill"]: len(node["requires"]) for node in res["dag"]}

# F2: map each step to its direct prerequisites' plan positions so the DAG edges

# (not just a count) are reconstructable from logs.

position_of = {node["id"]: i for i, node in enumerate(res["plan"], 1)}

requires_of = {node["skill"]: node["requires"] for node in res["dag"]}

emit.emit("skill_run", "plan_resolved", session_id, {

**base,

"recommended_products": ",".join(tokens),

"planned_skill_count": len(res["plan"]),

"dead_end_count": len(res["dead_ends"]),

"choice_count": len(res["choices"]),

}, critical=True)

for position, node in enumerate(res["plan"], 1):

deps = sorted(position_of[r] for r in requires_of.get(node["id"], [])

if r in position_of)

step = {

**base,

"step_kind": "skill",

"plan_position": position,

"skill_id": node["id"],

"skill_kind": node["kind"],

"product": node.get("product") or "",

"source_repo": (node.get("source") or {}).get("repo") or "",

"dependency_count": dependency_count.get(node["id"], 0),

"skill_invoked": False,

"instrumentation_invoked": False,

}

if deps: # omit for root steps (no edges)

step["depends_on"] = ",".join(str(p) for p in deps)

emit.emit("skill_step", "planned", session_id, step, critical=True)

# ponytail: one coverage_gap per dead-end, counted only. resolve.py's dead_ends are

# free text and privacy (proposal §8) forbids echoing them; splitting them into

# per-product tokens is a follow-up. dead_end_count (above) already carries the total.

for _dead_end in res["dead_ends"]:

emit.emit("skill_step", "planned", session_id, {

**base,

"step_kind": "coverage_gap",

"skill_invoked": False,

"instrumentation_invoked": False,

"result": "not_automated",

}, critical=True)

except Exception:

return

def main():

ap = argparse.ArgumentParser()

ap.add_argument("--products", default="")

ap.add_argument("--platform", default="none")

ap.add_argument("--cloud", default="none")

ap.add_argument(

"--select",

action="append",

default=[],

metavar="PRODUCT=SKILL_ID",

help="resolve an implementation choice (repeatable)",

)

ap.add_argument(

"--include-disabled",

action="store_true",

help="route over the full capability model, including disabled (private / not-GA) "

"implementations; shows the path a skill would take once its source is enabled",

)

ap.add_argument(

"--debug",

action="store_true",

help="also render the ASCII DAG of the skills to be executed",

)

ap.add_argument(

"--trace",

action="store_true",

help="emit a stable, machine-readable TRACE block (STOP_REASON, PLAN, DEAD_ENDS, "

"CHOICE_POINTS, CONFIRMED, DISPATCHED) for the run's trace file instead of the human table",

)

ap.add_argument(

"--list-products",

action="store_true",

help="print the accepted product vocabulary (recommender names + catalog tokens) and exit",

)

ap.add_argument(

"--detect-products",

default=None,

metavar="INTENT",

help="scan a free-text intent for explicitly named products; print the matched product "

"tokens (one CSV line, empty if none) and exit. Empty output => run the recommender",

)

ap.add_argument(

"--agent-name",

default="unknown",

help="executor agent for telemetry: claude_code | codex | cursor | unknown",

)

ap.add_argument(

"--intent-mode",

choices=("explicit", "recommended"),

default=None,

help="product provenance (REQUIRED with --trace): 'explicit' if the intent named the products "

"(--detect-products shortcut), 'recommended' if dd-product-recommender produced them. The "

"orchestrator may not compose a plan from products it inferred itself.",

)

a = ap.parse_args()

selections = {}

for item in a.select:

if "=" not in item:

ap.error(f"--select must be PRODUCT=SKILL_ID, got {item!r}")

product, skill_id = item.split("=", 1)

if not product.strip() or not skill_id.strip():

ap.error(f"--select must be PRODUCT=SKILL_ID, got {item!r}")

selections[product.strip()] = skill_id.strip()

catalog = load_catalog()

if a.list_products:

router = Router(catalog, enabled_only=not a.include_disabled)

print("Accepted --products inputs (case-insensitive, whitespace-tolerant):")

print(" names/aliases: " + ", ".join(sorted(PRODUCT_TOKENS)))

print(" catalog tokens: " + ", ".join(sorted(router.product_tokens)))

return 0

if a.detect_products is not None:

# Shortcut: the intent may already name products. If so, skip the recommender and

# route them directly; empty output tells the runbook to recommend instead.

router = Router(catalog, enabled_only=not a.include_disabled)

print(",".join(detect_products(a.detect_products, router.product_tokens)))

return 0

# Gate: a composed trace must declare product provenance — the --detect-products shortcut or

# dd-product-recommender. The orchestrator may not build a plan from products it inferred itself.

if a.trace and a.products.strip() and not a.intent_mode:

ap.error("--trace with --products requires --intent-mode explicit|recommended: products must "

"come from the --detect-products shortcut or dd-product-recommender, never inferred "

"by the orchestrator")

# One session id per invocation, shared by every event in this DAG run. Minted here

# (or taken from the environment if an outer wrapper already set it) and printed so the

# SKILL.md runbook can reuse it on each dispatch-boundary emit.py call.

session_id = os.environ.get("DD_ORCH_SESSION_ID") or str(uuid.uuid4())

if not a.trace:

# In --trace mode the framed block already carries SESSION_ID; keep this human-oriented

# line for plain runs only, so the trace stays a single machine-readable block.

print(f"SESSION ID: {session_id}")

products = [p for p in a.products.split(",") if p.strip()]

router = Router(catalog, enabled_only=not a.include_disabled)

res = router.resolve(

products,

{"platform": a.platform, "cloud": a.cloud},

selections=selections,

)

if a.trace:

_print_trace(session_id, a, res)

else:

_print(res)

if a.debug:

_print_dag(res)

_emit_telemetry(session_id, a, products, router, res)

return 0

if __name__ == "__main__":

sys.exit(main())