Dd Orchestrator
DatadogOfficialPublished by Datadog
- Skill
- Meta (authoring, orchestration, planning, memory)
- Automation and integrations
- MIT
- Low risk
- Has scripts
AutoGineer summary
Written by AutoGineer from the skill’s own files, not by the author
dd-orchestrator acts as the conductor for Datadog onboarding. It takes a user's plain-language goal, identifies the necessary Datadog products, detects the project's platform and cloud environment, and generates an ordered execution plan. It then dispatches this plan to individual, specialized skills, ensuring prerequisites are met and dependencies are respected. It is designed to be a non-performing orchestrator that relies on a catalog-driven approach to ensure consistent and reliable onboarding.
What it does
- Authenticates and validates Datadog account credentials via dd-account-setup.
- Determines required Datadog products based on user goals or codebase analysis.
- Detects project platform and cloud infrastructure to tailor the onboarding plan.
- Composes an ordered, dependency-aware execution plan using a central catalog.
- Dispatches tasks to specific skills, fetching them from source if necessary.
- Provides a structured summary of actions taken, gaps identified, and verification steps.
When to use it
- When a user requests to set up Datadog for their application.
- When a user wants to instrument a project without knowing specific product names.
- When onboarding a repository to Datadog for the first time.
- When a monitoring goal is stated but the specific Datadog products are not defined.
Advantages
- Ensures a consistent, repeatable onboarding process across different projects.
- Automates the complex dependency management between Datadog products and infrastructure.
- Provides clear visibility into the onboarding plan before any changes are applied.
- Identifies coverage gaps where no automated skill currently exists.
- Maintains a clean separation between orchestration logic and individual skill execution.
Usage examples
- “Set up Datadog for my app”
- The orchestrator will prompt for account setup, detect the app's platform, recommend relevant products, and present an ordered plan for approval.
- “I want to monitor my LLM chatbot in production”
- The orchestrator will identify LLM monitoring needs, check for required infrastructure, and propose a plan to enable relevant observability tools.
- “Instrument my React + Node app and catch frontend errors”
- The orchestrator will detect the frontend/backend stack and generate a plan to enable RUM and APM for error tracking.
Install
Download the skill as a ZIP:
https://autogineer.com/agent-skills/datadog-labs/dd-orchestrator/downloadUnzip it into the cross-client skills folder of your project:
mkdir -p .agents/skills && unzip dd-orchestrator.zip -d .agents/skills
Documentation by Datadog
Datadog Onboarding Orchestrator
You are the entry point for Datadog onboarding — the conductor, not a performer. You take one plain-language goal, decide which existing skills are needed and in what order, and hand off to them across all sources. You do not write instrumentation yourself; each skill owns its steps.
Routing is compositional, not a lookup. A product like "APM" is not one skill — it expands into
ensure account → install the Agent for the detected platform → enable the product → verify, and
surfaces any relevant cloud integration as an optional suggestion (never a required step). That
composition is computed from the capability graph in
catalog.json; there is no intent-to-skill table anywhere (intents live only in the recommender).
Ground rules (read once)
Confirm before you run. Show the composed plan (skills, in order) and the skipped dead-ends, and get a yes before dispatching anything — with one exception:
dd-account-setupruns first as a preflight (Step 2), because there is no plan to show until an account exists. Nothing else dispatches before approval.The catalog is the source of truth.
catalog.jsonholds every skill as a node with facets (kind,product,platform,cloud), a category-levelrequiresgraph, and asource.url.resolve.pycomposes the plan from it. Do not hand-maintain a routing table.Only enabled skills route. Each node carries
enabled(the two public sets —agent-skillsdd-sourcestatus: ga— plus this repo's own skills; everything else is disabled).resolve.pycomposes only from enabled skills in real time; a disabled skill is treated as unavailable and surfaces as a dead-end (demand signal).
Use repository evidence and the developer's stated context. Platform (kubernetes/docker/lambda/host…) and cloud (aws/gcp/…) can come from the repository or an unambiguous statement in the goal. For example, "my Kubernetes service on AWS" supplies
platform=kubernetesandcloud=aws. Ask only when neither source determines the needed value, or when they conflict. Do not guess an unstated value or re-ask for one that is already clear.Conduct, don't perform. You compose and dispatch skills; you never do a delegated skill's job, pre-empt its decisions, or turn them into a user choice. In particular: product selection is
dd-product-recommender's job — never infer, guess, or shortlist products yourself, and never offer product scope as a choice; and authentication + site/region aredd-account-setup's job — invoke it and let it ask. The only choices you surface concern missing or conflicting structural context (platform/cloud), including the CHOICE POINTS thatresolve.pyemits.Never fabricate a skill or a step; dispatch only what the resolver planned. If a recommended product has no node, or is not covered for the detected platform, say so plainly, point to
docs.datadoghq.com, and record it as a gap. Two hard gates apply before any dispatch:No plan, no dispatch. If you have not run
resolve.pyand captured itsPLANblock (with theSESSION ID) for this run, dispatch nothing — the resolver plan is the only dispatch authority. Do not hand-build a plan.Every id must be in the PLAN. Before you dispatch a skill id, confirm it appears in that
PLANblock (e.g.grepthe id in the run's trace file — see Step 5). An id not in the PLAN is fabricated: do not dispatch it; record it as a gap.
Account first. Nothing routes before
dd-account-setupreports a valid key on the right region.Invoke every planned skill — installed or not; never skip, never substitute. Dispatch each skill once, in the resolved order, from its recorded
sourceincatalog.json. If the skill id is in the session registry, invoke it directly. If it is not, fetch it from source and run it inline (invoke, not install):python3 dd-orchestrator/scripts/fetch_skill.py <id>materializes the skill (itsSKILL.mdplus anyreferences/andscripts/) into a temp dir from the newest public source, and you then execute thatSKILL.md. Sources always track the newest version (main/ the live onboarding-API render) — nothing is pinned. You may NOT infer, summarize, or hand-author a skill's result in its place. The only permitted non-execution is a hard failure of the fetch or the skill itself, which stops that dependency chain and is reported (see Sequential dispatch below) — never a silent skip. Iffetch_skill.pyexits non-zero (for example, HTTP 404 or a blocked/non-allowlisted source), record that node underDEAD_ENDSwith the actual fetch error and mark it not-run. Stop its dependency chain. Do not substitute a local, cached, vendored, or on-disk copy after a failed fetch, put it underDISPATCHED, or claim the product is set up.Sequential dispatch — chains first, stop on failure. Run the plan strictly in
resolve.py's order (it is deterministic and independent of the product request order, and keeps each dependency chain contiguous). Execute one skill at a time; do not start a skill until every hard prerequisite has succeeded, not merely been dispatched. Each skill runs at most once. If a skill fails, stop that chain: skip its transitive dependents and report them as not-run — an independent branch is unaffected.Checklist discipline. Post a checklist up front and tick items as you go (per this repo's CLAUDE.md).
Output discipline — quiet by default. The only user-facing output is: the checklist (post it once, then tick items in place — do not reprint it), the PLAN block, any choice questions, and the SUMMARY block. Between the PLAN block and the SUMMARY block, write at most one short status line per plan node (e.g.
-> dd-account-setup ...thendone: dd-account-setup). Do not narrate your reasoning, read files aloud, restate the goal or the plan, or re-explain these rules. The audit detail belongs in the run's trace file (Step 5) — point to it; do not reprint it. Keep your reasoning internal; do not think out loud. Setup and telemetry plumbing — capturing the org id, running preflight shell commands, emittingemit.pyevents — is internal: run it silently and never announce it or report its outcome (e.g. "capturing the org id", "org id not captured"), unless the context explicitly asks for debug info. Fewer words, no lost meaning (this pairs with Simplified Technical English below).Correctness first; styling and telemetry are best-effort. The load-bearing path is: account first → compose with
resolve.py→ confirm → dispatch each planned skill once, in order. Per-step telemetry (emit.py) and the Simplified Technical English styling below are best-effort. If you are under load, on a tight budget, or a telemetry call is failing, skip them and keep going — they never change the plan, the dispatch, or the grade.Prefer Simplified Technical English (ASD-STE100) (best-effort — see Correctness first above). All user-facing output — the checklist, the plan, choice questions, gaps, and the summary — uses Simplified Technical English (ASD-STE100). Apply its core rules:
Keep sentences short: at most 20 words for an instruction, 25 for a description. Write one instruction per sentence.
Use the active voice, the imperative for instructions, and simple verb tenses. Avoid
-ingforms where a simpler verb works.Keep the articles ("a", "the") and write complete sentences. Do not use a telegraphic or headline style.
Use one word for one meaning, and keep the same term for the same thing. Avoid slang, jargon, and undefined abbreviations.
Prefer short, common words. Use lists for steps, and keep paragraphs short. The result is short but complete: fewer words, no lost meaning.
The flow
developer goal
│
├─► dd-account-setup precondition: valid key, right region
├─► dd-product-recommender goal + codebase → ranked PRODUCTS (skipped if the intent names products)
├─► detect context platform + cloud from the repo or an unambiguous goal
│
▼
scripts/resolve.py --products "<recommended>" --platform <detected> --cloud <detected>
│ (reads catalog.json: binds category requires to detected context,
│ orders hard edges depth-first for dependency-chain locality
│ (kind breaks ties), dedupes, appends verify)
▼
PLAN (ordered skills) + DEAD-ENDS (recorded) + CHOICE POINTS (ask)
│
▼
confirm → dispatch each plan node by source.url → summarize (incl. the skipped gaps)Steps
Post the checklist.
Ensure the account — invoke
dd-account-setup(installed → invoke directly; not installed → fetch it from source and run it inline, per Ground rules); stop if it cannot produce a validated key.dd-account-setupowns the site/region and authentication prompts — do not pre-empt them; invoke it and let it ask. Then capture the authenticated org id for telemetry (best-effort — any failure just leaves it unset and the field is omitted). This works on any validated path: prefer the OAuth Bearer token thatdd-account-setupleaves in place, and fall back to the API+APP key pair — so an OAuth sign-in that has no app key still resolves the org (do not try to mint an app key just for this). Run once, before Step 5, and silently — do not announce the capture or its outcome (it is best-effort telemetry); surface it only if the context asks for debug info:tf="${TMPDIR:-/tmp}/dd-oauth-$(id -u).token" if [ -s "$tf" ]; then hdr=(-H "Authorization: Bearer $(cat "$tf")") elif [ -n "$DD_APP_KEY" ]; then hdr=(-H "DD-API-KEY: $DD_API_KEY" -H "DD-APPLICATION-KEY: $DD_APP_KEY") else hdr=(); fi [ ${#hdr[@]} -gt 0 ] && export DD_ORG_ID="$(curl -sf -m 5 "${hdr[@]}" \ "https://api.${DD_SITE:-datadoghq.com}/api/v2/current_user" \ | python3 -c 'import sys,json; d=json.load(sys.stdin); o=d["data"]["relationships"]["org"]["data"]["id"]; print(next((x["attributes"]["public_id"] for x in d.get("included",[]) if x.get("type")=="orgs" and x.get("id")==o), o))' 2>/dev/null || true)"resolve.py(Step 5) readsDD_ORG_IDinto the run envelope, so every event carries the org id.Products — shortcut or recommend. First check whether the intent already names products:
python3 dd-orchestrator/scripts/resolve.py --detect-products "<intent>".If it prints one or more product tokens, the user named the products: use that list and skip the recommender. (e.g. "Instrument RUM and LLMO" →
rum,llm-obs.)If it prints nothing, the intent describes a goal: you MUST invoke
dd-product-recommenderand use its ranked product list (installed → invoke directly; not installed → fetch it from source and run it inline, per Ground rules). Never infer, guess, or shortlist products yourself from the stack/framework, and never offer product scope as a user choice — product selection is the recommender's job, and running it is mandatory here. (e.g. "Help me track user actions" → recommend.) The shortcut skips only the recommendation. Account setup (Step 2), context detection (Step 4), the confirm gate, andresolve.pystill run — the shortcut is never a bypass of a safety gate. No legitimate products, noresolve.py. The product list passed to Step 5 MUST come from exactly one of: the--detect-productsshortcut (→--intent-mode explicit) ordd-product-recommender(→--intent-mode recommended). You may not compose or preview a plan from products you authored yourself;resolve.py --tracerefuses to run without a declared--intent-mode.
Detect context — use repository evidence (k8s manifests, Dockerfile,
serverless.yml, Terraform/SDK/provider signals) or an unambiguous platform/cloud stated in the goal. Pass known values toresolve.pyas--platform/--cloud. Leave unknown values unset; ask when a needed value is missing or the repository and goal conflict. Do not replace an explicitly stated platform withnonejust because the repository lacks deployment files.Compose the plan and seed the trace. The trace is a run-scoped scratch file the orchestrator owns —
${DD_ORCH_OUTPUT_DIR:-${TMPDIR:-/tmp}/dd-orchestrator}/trace.md— not a bareoutput/in the user's project (a relative path is cwd-dependent and could overwrite the user's own files). The default is underTMPDIR, so it never litters the repo and is cleaned automatically; setDD_ORCH_OUTPUT_DIRto override it (the eval points it at its workspace). Reset the scratch — safe, because the path is the orchestrator's own namespace, not a guessedoutput/— then run:TRACE="${DD_ORCH_OUTPUT_DIR:-${TMPDIR:-/tmp}/dd-orchestrator}/trace.md"; mkdir -p "$(dirname "$TRACE")" && rm -f "$TRACE"python3 dd-orchestrator/scripts/resolve.py --trace --products "<products>" --platform <platform> --cloud <cloud> --intent-mode <explicit|recommended> | tee "$TRACE"Pass--intent-mode explicitwhen Step 3's shortcut named the products, or--intent-mode recommendedwhendd-product-recommenderproduced them. It is required —resolve.py --tracerefuses to compose a plan without it (the products must trace to the shortcut or the recommender, never to your own inference) — but it never changes the plan itself. The--traceflag prints one stable, machine-readable block —SESSION_ID,STOP_REASON,PLAN,DEAD_ENDS,CHOICE_POINTS,SUGGESTED,CONFIRMED,DISPATCHED— andteesaves it verbatim to$TRACE. That deterministic block IS your dispatch trace; never re-narrate the plan by hand. Read the same output for the ordered plan, the dead-ends, and any choice points. Capture theSESSION_ID:value from that block in a run-local variable:SID=<SESSION_ID>. Pass--session-id "$SID"to every laterresolve.pyre-run andemit.pycall. Capture a fresh value from the first trace of each new run; do not exportDD_ORCH_SESSION_ID. If an outer wrapper supplies that environment variable, the resolver honors it; leave its value unchanged. A new id during a replan would split one run's telemetry and restart itsevent_seq. In debug mode (only when the context explicitly asks for it), add--debugto also render the ASCII DAG (indent = dependency depth,<-= direct prerequisites).resolve.pyalso emits the reliable telemetry core here (see Telemetry below). After seeding the trace, record the successfuldd-account-setuppreflight as the first line underDISPATCHED. Restore that entry after any choice-driven trace regeneration, including when no further skills can run.Resolve choices — for each choice point (e.g. "pick a platform: kubernetes, linux"), ask the developer and re-run with the confirmed value and
--session-id "$SID"from Step 5 so the updated plan and subsequent telemetry remain part of the same run.Confirm — render the PLAN block, then dispatch. Before any dispatch, render the PLAN block (see Output templates below) verbatim: fill the slots, add no extra prose, keep the exact section order and headers. The Plan table's Source column is the skill's source URL, as
resolve.pyprints it (the newest public source;selffor this repo). On approval, run each plan node in dispatch order, one at a time — invoking each skill (installed → directly; not installed → fetched from source and run inline, per Ground rules) — and ticking the checklist. Whenever the plan begins withdd-account-setup(every plan that needs an account — i.e. any non-empty onboarding plan), Step 2 already ran it: do not invoke it a second time; tick that node, retain its firstDISPATCHEDentry, and emit itsskill_step:started/finishedfrom the preflight result, then continue with the next node. Do not start a node until its prerequisites succeeded; if one fails, skip its dependents. As you dispatch, emit best-effort per-step telemetry (Telemetry below):skill_step:startedbefore a node, thenskill_step:finished(withresult+duration_ms) orskill_step:skipped. Keep the trace file ($TRACE, Step 5) current: setCONFIRMED: yeson approval, orCONFIRMED: noif the user declines. Add eachskill_idtoDISPATCHEDonce, in execution order, only after it ran successfully — directly if installed, or fetched-and-run otherwise. A failed fetch or execution belongs underDEAD_ENDSwith its actual error; distinguish "not run" from "ran and failed", and stop that dependency chain.DISPATCHEDrecords successful executions, starting with the account preflight. If choices remain, the plan is empty, or the user declines, run no further skills; keep the successful preflight entry without inventing additional dispatches.Summarize — render the SUMMARY block. As the terminal output, render the SUMMARY block (see Output templates below) verbatim, in the exact section order. The skipped / gaps list is a real output — the coverage-gap / demand signal for what to automate next; state it, do not hide it. Emit one terminal
skill_run:finished(Telemetry).
Output templates
Render two fixed blocks so every run reads the same: the PLAN block at Step 7 (the confirm gate) and the SUMMARY block at Step 8 (terminal output). Fill the slots and add no extra prose.
Rendering rules (both blocks). Use the given section order and the exact headers. Prefer tables to
prose; one line per row; do not editorialize or restate the goal. Every dispatched step shows its
source (the skill's public source URL — newest; self for this repo). Action items are imperative and
carry the exact command or file path. Always end with the telemetry session_id so the output and the
events join. Omit a section only by its stated omission rule. Marker legend: ✓ done/verified · ◑
partial or wired-not-verified · ⚠ needs action / mutating · ✗ failed/blocked · ⊘
skipped/not-covered.
PLAN block — Step 7 (before any dispatch)
# Datadog Onboarding — Plan · session {{session_id}}
## Detected context
- Platform: {{platform}} · Cloud: {{cloud|none}} · Stack: {{stack_summary}}
- Existing Datadog: {{existing|none}}
## Recommended products
{{i}}. {{product}} · {{priority}} · {{one-line why, names a file/lib}}
## Plan — {{k}} step(s), in dispatch order
| # | Skill | Kind | Product | Source |
|---|-------|------|---------|--------|
| {{n}} | {{skill_id}} | {{kind}} | {{product}} | {{source url (newest) or self}} |
Dependencies: {{root}} → {{chain / branches, one line}}
## Not automated ({{dead_end_count}})
- {{product}} — {{why}} → {{docs URL}}
> Show "None — every recommended product is covered." when dead_end_count = 0.
## Decisions needed ({{choice_count}})
- {{choice}}: {{optionA}} / {{optionB}} / {{optionC}}
> Show "None." when choice_count = 0.
## Before you approve — effects
- ⚠ {{step}} {{mutating / outward-facing effect}}
- {{step}} {{non-mutating effect}}
Approve? [Proceed — all {{k}}] · [Cancel]SUMMARY block — Step 8 (terminal output)
# Datadog Onboarding — Summary · session {{session_id}} · {{result}}
## Checklist
- [{{x|.}}] #{{n}} {{skill_id}} — {{one-line outcome}}
## Products set up
| Product | Delivered by | Status | Evidence |
|---------|--------------|--------|----------|
| {{product}} | {{mechanism, source}} | {{✓|◑|✗}} | {{proof or "pending {{blocker}}"}} |
## Changed
- Cluster: {{namespaces/resources}}
- App: {{files/manifests}}
- Creds: {{where, gitignored?}}
## Action items — do next
1. [ ] {{imperative}} — `{{exact command / path}}`
> Show "None — setup is complete." when there are no follow-ups.
## Issues & deviations
| What | Cause | Resolution / impact |
|------|-------|---------------------|
| {{issue}} | {{cause}} | {{how resolved / residual impact}} |
> Show "None." when the run was clean.
## Gaps / demand signals
- {{skipped product or catalog/orchestrator gap}}
## Verify in Datadog
- {{product}}: {{deep link}}
## Telemetry
session {{session_id}} · {{event_count}} events · result {{result}} ({{s}}✓ / {{f}}✗ / {{k}}⊘)Run verdict — the overall
{{result}}: usesuccesswhen every planned skill succeeded, even if some recommended products are dead-ends / "Not automated" (no skill yet) — those are coverage gaps, not partial failures. Usepartial_successwhen a planned fetch or execution failed, or a dependent step was skipped; do not infer success fromDISPATCHEDalone. Useblockedwhen nothing beyond the account preflight can run because every product dead-ended;cancelledwhen the user declined the plan. Report failures and skips in the telemetry counts even though those skills are absent fromDISPATCHED;emit.pyreconciles the verdict from those counts.
Telemetry (best-effort — never blocks onboarding)
All telemetry goes through emit.py; never build your own HTTP request or curl. It is
best-effort by construction (bounded timeout, local debug log, never throws) and emits to the
logs-intake route only. Turn it off with DD_ORCH_TELEMETRY_DISABLED=1. Reuse the single
SID captured in Step 5 via --session-id "$SID" on every call so the whole run stitches together.
resolve.py already emits the reliable core: one skill_run:started per session, then
skill_run:plan_resolved and one skill_step:planned per plan node and per dead-end on each resolve
(each skill_step also carries depends_on
— the CSV of prerequisite plan positions — so the DAG edges are reconstructable). resolve.py
persists the run envelope (agent, platform, cloud, entry, intent mode, org id) and emit.py re-attaches it plus an
emitted_at (ms) timestamp to every event automatically — so you need not re-pass the envelope;
send only the per-step fields below. During dispatch you add the per-step lifecycle and the terminal
run event:
# SID is the run-local SESSION_ID captured in Step 5.
# before invoking a plan node (source_mode records how it ran: installed vs fetched-from-source)
python3 dd-orchestrator/scripts/emit.py skill_step --action started --session-id "$SID" \
--field plan_position=<n> --field skill_id=<id> --field skill_kind=<kind> \
--field product=<product> --field source_repo=<repo> --field source_mode=<installed|fetched>
# after it returns
python3 dd-orchestrator/scripts/emit.py skill_step --action finished --session-id "$SID" \
--field plan_position=<n> --field skill_id=<id> --field result=success \
--field duration_ms=<ms> --field skill_invoked=true \
--field instrumentation_invoked=<true if it was an install/connect/enable skill>
# if a node is NOT run (failed prerequisite, user declined, no automation, source unreachable)
python3 dd-orchestrator/scripts/emit.py skill_step --action skipped --session-id "$SID" \
--field plan_position=<n> --field skill_id=<id> --field result=skipped_dependency
# once, when the run reaches a terminal state
python3 dd-orchestrator/scripts/emit.py skill_run --action finished --session-id "$SID" \
--field result=<success|partial_success|failed|blocked|cancelled> \
--field step_success_count=<n> --field step_failed_count=<n> --field step_skipped_count=<n>Field values are bounded enums / ids / counts only — never send goal text, paths, keys, URLs, or model output (the emitter also strips anything not on its allow-list).
Reliability and reconciliation. Every event carries a per-session event_seq (a monotonic
ordinal): a gap in event_seq means an event was dropped, not that the step never ran. resolve.py
emits its plan-shape core (skill_run:started, skill_run:plan_resolved, one skill_step:planned
per node) as critical — one transport blip cannot drop the whole core. emit.py also appends
every attempted event to a durable local NDJSON log for offline audit. When you analyze a run, treat
step_success_count / step_failed_count / step_skipped_count on skill_run:finished as the
source of truth for how many steps ran; reconcile the per-step events against it. Do not assume a
missing per-step event means the step did not run. emit.py also reconciles the terminal
skill_run:finished result against those counts (and the dead-end count): if the reported verdict
contradicts them, it keeps the reported value as result_reported and sets result to the
count-consistent verdict (result_reconciled: true). So report the honest per-step results and let the
guard settle the run verdict.
Worked example
Goal "monitor my Node service on Kubernetes" → recommender [APM, Infrastructure Monitoring],
detected platform=kubernetes:
1. dd-account-setup (foundation) [self]
2. apm-agent-install-kubernetes (platform-install) [agent-skills]
3. apm-enable-kubernetes (product-enable/apm) [agent-skills]
4. apm-verify-ssi-kubernetes (verify) [agent-skills]One product plus a detected platform became a four-skill plan drawn from two repos, correctly ordered, with Infrastructure Monitoring delivered by the same SSI Agent install. The intent never entered the resolver — only the products and the detected platform did.
Files
- ATTRIBUTION.mdAdded by AutoGineer866 B
- LICENSELicence file added by AutoGineer1.1 KiB
- SKILL.md26.3 KiB
- catalog.json53.9 KiB
- scripts/emit.pyScript19.2 KiB
- scripts/fetch_skill.pyScript5.6 KiB
- scripts/resolve.pyScript47.1 KiB
scripts/emit.py
#!/usr/bin/env python3
"""Best-effort telemetry emitter for dd-orchestrator (v1).
One tiny emitter for the whole skill. Skills NEVER build their own HTTP request or
`curl` — they call this. It builds the setup-cli logs-intake log line and POSTs it to
Datadog logs intake, best-effort:
* bounded 3s timeout,
* retry ONLY transient 5xx (never 429),
* a 1-strike per-process breaker so a down intake can't stall a multi-event run,
* an append-only local debug log,
* and it never raises or blocks the caller.
v1 uses the logs-intake route only (no signup route), so it has no cross-repo
dependency. It reuses setup-cli's publishable ``pub…`` client-token intake, which
lands events in the same internal ``service:setup-cli`` pipeline the onboarding-growth
team already consumes. It deliberately does NOT use the customer's DD_API_KEY: that
would write onboarding telemetry into the customer's own org, not Datadog's.
CLI (the runbook calls this at each dispatch boundary — see SKILL.md):
python3 emit.py skill_run --action started --session-id <uuid> [--field k=v ...]
python3 emit.py skill_step --action started --session-id <uuid> --field skill_id=<id> ...
python3 emit.py skill_run --action finished --session-id <uuid> --field result=success ...
The session id is minted and printed by resolve.py ("SESSION ID: <uuid>"); every
event in one DAG run must reuse it.
"""
from __future__ import annotations
import argparse
import datetime
import json
import os
import sys
import tempfile
import time
import urllib.error
import urllib.request
from urllib.parse import quote
SCHEMA_VERSION = 1
# --- envelope the runtime owns (a caller can NEVER override these) -----------
RUNTIME_SOURCE = "setup-cli" # reuse the existing pipeline -> service:setup-cli
FLOW = "agent"
PLATFORM = "cli"
EVENT_TYPES = {"skill_run", "skill_step"}
ACTIONS = {
"skill_run": {"started", "plan_resolved", "finished"},
"skill_step": {"planned", "started", "finished", "skipped"},
}
# Reconciled with catalog.json `kinds`: a `lifecycle` node CAN enter a plan, so it is
# a valid skill_kind (the proposal's enum omitted it). `internal` is filtered by
# resolve.py and never reaches a step, so it is intentionally absent here.
SKILL_KINDS = {"foundation", "platform-install", "cloud-connect",
"product-enable", "verify-troubleshoot", "lifecycle"}
# --- client-side allow-list (mirrors the server-side allow-list intent) ------
# Any key not here is stripped before send (privacy + bounded cardinality). Grouped
# only for readability; the union is what gates.
_FACETS = {
"event_type", "event_action", "result", "invocation_mode", "intent_mode", "entry_skill_id",
"agent_name", "headless", "target_platform", "target_cloud", "step_kind",
"skill_id", "skill_kind", "product", "source_repo", "source_mode", "error_code", "result_reported",
}
_MEASURES = {
"duration_ms", "plan_position", "dependency_count",
"planned_skill_count", "dead_end_count", "choice_count",
"step_success_count", "step_failed_count", "step_skipped_count",
}
_ATTRIBUTES = {
"schema_version", "session_id", "source", "flow", "platform",
"agent_version", "recommended_products", "emitted_at", "depends_on", "event_seq",
"skill_invoked", "instrumentation_invoked", "result_reconciled",
"org_id", # authenticated org's public_id, captured once at auth; attribute (not a facet)
}
ALLOWED_KEYS = _FACETS | _MEASURES | _ATTRIBUTES
_BOOL_KEYS = {"headless", "skill_invoked", "instrumentation_invoked", "result_reconciled"}
# setup-cli's publishable logs-intake client tokens (`pub…` = write-only, meant to
# ship in clients). Kept in sync with setup-cli/src/constants/credentials.ts. An env
# override always wins.
_CLIENT_TOKENS = {
"datadoghq.com": "pub5b34c05a3abc178e4d2abddf11bb4f31",
"us3.datadoghq.com": "pub82dbd99bf6299230cee3685535d4660d",
"us5.datadoghq.com": "pubf9a4a5b93cc00e7f4d5628f05ba1e7f0",
"datadoghq.eu": "pub471980dfdfb12f9cf2ffdf83551c9cb7",
"ap1.datadoghq.com": "pubbbc10e19ab9fa2f7eb54250610e9bbad",
"ap2.datadoghq.com": "pube391858a603cb55a0e72a8cb9f9fb627",
"uk1.datadoghq.com": "pub90fe05f76003a728ed554865d39a88d1",
}
_TAGS = "service:setup-cli"
# Per-process breaker. Non-critical emits are 1-strike (a down intake can't add N*3s to a
# run). CRITICAL emits (resolve.py's plan-shape core) BYPASS an open breaker so a single
# transient blip cannot drop run:started + plan_resolved + planned×N; a critical failure
# trips the breaker only after 3 consecutive strikes (bounds the worst-case hang).
_BREAKER = {"open": False, "critical_fails": 0}
def _trip_breaker(critical):
"""Record one terminal failure (transport OR terminal HTTP) against the breaker.
Non-critical: a single strike opens it. Critical (resolve-core): opens only after
3 consecutive failures, so one blip cannot drop the plan shape — but once that
bound is reached the breaker applies to critical events too (see ``emit``)."""
if critical:
_BREAKER["critical_fails"] += 1
if _BREAKER["critical_fails"] >= 3: # bound the worst-case hang
_BREAKER["open"] = True
else:
_BREAKER["open"] = True # non-critical: 1-strike
def _truthy(value):
return str(value).strip().lower() in ("1", "true", "yes", "on")
def _is_disabled():
if _truthy(os.environ.get("DD_ORCH_TELEMETRY_DISABLED", "")):
return True
# ponytail: never emit real telemetry from a test run — the pub token writes to the
# internal setup-cli index. A genuine CLI / orchestrator run never imports these.
if "unittest" in sys.modules or "pytest" in sys.modules:
return True
return False
def _debug_log_path():
return os.environ.get(
"DD_ORCH_TELEMETRY_LOG",
os.path.join(tempfile.gettempdir(), "dd-orchestrator-telemetry.log"),
)
def _debug(msg):
"""Append one line to the local debug log. Never raises."""
try:
with open(_debug_log_path(), "a") as handle:
handle.write(msg.rstrip("\n") + "\n")
except Exception:
pass
def _coerce(key, value):
if key in _BOOL_KEYS:
return value if isinstance(value, bool) else _truthy(value)
if key in _MEASURES:
try:
return int(value)
except (TypeError, ValueError):
return None # drop an unparseable measure rather than send a bad type
return value
# --- run-scoped state: shared by sequential resolve.py and emit.py processes ----------
# Replans refresh the envelope and shape while preserving lifecycle state and event_seq.
_ENVELOPE_KEYS = ("entry_skill_id", "invocation_mode", "intent_mode", "agent_name",
"target_platform", "target_cloud", "org_id")
def _session_state_path(session_id):
safe = "".join(c for c in str(session_id) if c.isalnum() or c in "-_")
return os.path.join(tempfile.gettempdir(), f"dd-orch-{safe}.json")
def _load_state(session_id):
try:
with open(_session_state_path(session_id)) as fh:
state = json.load(fh)
return state if isinstance(state, dict) else {}
except Exception:
return {}
def write_session_state(session_id, envelope, shape=None):
"""Persist the run's envelope (re-attached to every later emit) and, when given, the plan
shape (`dead_end_count`, ...), so the terminal skill_run:finished can reconcile `result`
against the real counts. Preserves the seq counter across re-writes. Never raises."""
try:
if not session_id:
return
state = _load_state(session_id)
state["envelope"] = {k: (envelope or {})[k] for k in _ENVELOPE_KEYS
if (envelope or {}).get(k) not in (None, "")}
if shape is not None:
state["shape"] = {k: int(v) for k, v in shape.items()}
state.setdefault("seq", 0)
with open(_session_state_path(session_id), "w") as fh:
json.dump(state, fh)
except Exception:
pass
def _read_session_envelope(session_id):
env = _load_state(session_id).get("envelope", {})
return {k: v for k, v in env.items() if k in _ENVELOPE_KEYS and v not in (None, "")}
def _claim_run_start(session_id):
"""Reserve one run-start attempt across sequential processes, before transport.
Keep the marker even if transport fails: the local event log records that attempt.
If state cannot be persisted, allow emission so telemetry remains best-effort.
"""
try:
state = _load_state(session_id)
if state.get("run_started"):
return False
state["run_started"] = True
with open(_session_state_path(session_id), "w") as fh:
json.dump(state, fh)
return True
except Exception:
return True
def _next_seq(session_id):
"""Return the next per-session monotonic ordinal (1-based). A missing ordinal on the
export means an event was dropped, not that the step never ran (gap detection). Never
raises; returns None if state is unwritable (then event_seq is simply omitted).
ponytail: no lock — dispatch is sequential (one emit.py process at a time)."""
try:
state = _load_state(session_id)
seq = int(state.get("seq", 0)) + 1
state["seq"] = seq
state.setdefault("envelope", {})
with open(_session_state_path(session_id), "w") as fh:
json.dump(state, fh)
return seq
except Exception:
return None
def _ndjson_path():
return os.environ.get(
"DD_ORCH_TELEMETRY_NDJSON",
os.path.join(tempfile.gettempdir(), "dd-orchestrator-telemetry.ndjson"),
)
def _record_ndjson(event, status):
"""Append one durable NDJSON line per ATTEMPTED event so drops are auditable/replayable
offline. Never raises."""
try:
with open(_ndjson_path(), "a") as fh:
fh.write(json.dumps({"status": status, "event": event}) + "\n")
except Exception:
pass
def _now_iso():
return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="milliseconds")
def _read_run_shape(session_id):
shape = _load_state(session_id).get("shape", {})
return shape if isinstance(shape, dict) else {}
# Verdicts we re-check against the counts. An agent-reported `blocked` / `cancelled` is trusted
# and never overridden; but a reported success/partial/failed that contradicts a pure opt-out
# (only skips) or all-dead-end count-shape is corrected TO `cancelled` / `blocked` — so a user
# who declined is not lumped into the failure funnel (PR #194 review). Both are SKILL.md verdicts.
_RECONCILABLE_RESULTS = ("success", "partial_success", "failed")
def _reconcile_run_result(out, session_id):
"""Guard the terminal run `result` against the real counts, so an agent verdict that
contradicts them (e.g. `partial_success` with 5 success / 0 failed / 0 skipped / 0 dead
ends) cannot skew the funnel. On disagreement, keep the agent's value as `result_reported`
and set `result` to the count-consistent verdict (`result_reconciled: true`)."""
result = out.get("result")
if result not in _RECONCILABLE_RESULTS:
return
if not any(k in out for k in ("step_success_count", "step_failed_count", "step_skipped_count")):
return # no counts on this event -> nothing to reconcile against
s = int(out.get("step_success_count", 0) or 0)
f = int(out.get("step_failed_count", 0) or 0)
k = int(out.get("step_skipped_count", 0) or 0)
d = int(_read_run_shape(session_id).get("dead_end_count", 0) or 0)
if s >= 1 and f == 0 and k == 0:
derived = "success" # everything dispatched succeeded; dead-ends are coverage
# gaps (a product with no skill yet), not partial failures
elif s >= 1:
derived = "partial_success" # something succeeded, but a step failed or was skipped
elif f == 0 and d == 0 and k >= 1:
derived = "cancelled" # nothing ran, nothing errored: user declined every step
elif f == 0 and d >= 1:
derived = "blocked" # nothing errored, but dead-ends stopped all progress
else:
derived = "failed" # something actually failed
if derived != result:
out["result_reported"] = result # preserve the agent's verdict
out["result"] = derived # count-consistent truth for the funnel
out["result_reconciled"] = True
def build_event(event_type, event_action, session_id, fields=None):
"""Return the validated event dict.
Caller `fields` are allow-list-stripped and type-coerced first; then the
runtime-owned envelope is applied ON TOP, so a caller can never spoof
session_id / source / event identity.
"""
out = {}
for key, value in (fields or {}).items():
if key not in ALLOWED_KEYS or value is None or value == "":
continue
coerced = _coerce(key, value)
if coerced is not None:
out[key] = coerced
# Re-attach the run-scoped envelope (persisted by resolve.py) so runbook emits that did
# not re-pass it still carry agent/platform/entry (F4). Runtime-owned -> un-spoofable.
out.update(_read_session_envelope(session_id))
if event_type == "skill_run" and event_action == "finished":
_reconcile_run_result(out, session_id) # `result` must match the step counts
out.update({
"event_type": event_type,
"event_action": event_action,
"schema_version": SCHEMA_VERSION,
"session_id": session_id,
"source": RUNTIME_SOURCE,
"flow": FLOW,
"platform": PLATFORM,
"emitted_at": _now_iso(), # F1: client ms timestamp -> reliable order + duration
})
return out
def _log_body(event):
"""The flat logs-intake log line (mirrors setup-cli's sendToIntake body)."""
return json.dumps({
"service": RUNTIME_SOURCE,
"message": f"{event['event_type']}:{event['event_action']}",
**event,
})
def _resolve_token(site):
override = os.environ.get("DD_ONBOARDING_CLIENT_TOKEN")
if override:
return override
return _CLIENT_TOKENS.get(site)
def _intake_url(site, token):
return (f"https://http-intake.logs.{site}/v1/input/{token}"
f"?ddsource=dd-orchestrator&ddtags={quote(_TAGS)}")
def _default_post(url, body):
"""POST the JSON body with a 3s timeout. Returns the HTTP status (int) or None on a
transport error / timeout. Never raises."""
request = urllib.request.Request(
url, data=body.encode("utf-8"), method="POST",
headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(request, timeout=3) as response:
return response.getcode()
except urllib.error.HTTPError as err:
return err.code
except Exception:
return None
def emit(event_type, event_action, session_id, fields=None,
*, site=None, disabled=None, _post=None, critical=False):
"""Best-effort emit of one event. Returns True iff intake accepted it. Never raises,
never blocks beyond the bounded timeout / retries.
``critical`` marks resolve.py's plan-shape core (run:started, plan_resolved, planned×N):
it BYPASSES an open breaker — but only until 3 consecutive critical failures trip it —
so one transient blip cannot drop the whole core, yet a persistently down intake still
stops after the bound instead of adding a timeout per plan node.
"""
try:
if disabled is None:
disabled = _is_disabled()
if disabled:
return False
if event_type not in EVENT_TYPES or event_action not in ACTIONS.get(event_type, ()):
_debug(f"drop: unknown event {event_type}:{event_action}")
return False
if not session_id:
_debug(f"drop: missing session_id for {event_type}:{event_action}")
return False
site = site or os.environ.get("DD_SITE") or "datadoghq.com"
token = _resolve_token(site)
if not token:
_debug(f"drop: no client token for site {site!r}")
return False
if event_type == "skill_run" and event_action == "started":
if not _claim_run_start(session_id):
return False
event = build_event(event_type, event_action, session_id, fields)
seq = _next_seq(session_id)
if seq is not None:
event["event_seq"] = seq # gap-detection ordinal
body = _log_body(event)
# Softened breaker: a critical (resolve-core) emit bypasses an open breaker so a
# single earlier blip cannot drop the plan shape — but only until the critical
# failure bound is reached; past that the breaker applies to critical events too,
# so a down intake cannot keep adding a timeout per plan node.
if _BREAKER["open"] and (not critical or _BREAKER["critical_fails"] >= 3):
_record_ndjson(event, "suppressed")
return False
url = _intake_url(site, token)
post = _post or _default_post
status = None
for attempt in range(3):
status = post(url, body)
if status is not None and 500 <= status <= 599: # retry transient 5xx only
_debug(f"retry {event_type}:{event_action} attempt {attempt + 1} status {status}")
time.sleep(0.2)
continue
break # 2xx, 429, other 4xx, or None: stop
_record_ndjson(event, status if status is not None else "transport_failed")
if status is None: # transport failure
_trip_breaker(critical)
_debug(f"transport-failed {event_type}:{event_action}")
return False
if status >= 400: # terminal HTTP failure (4xx/5xx/429)
_trip_breaker(critical) # count it toward the same bound
_debug(f"dropped {event_type}:{event_action} status {status}")
return False
_BREAKER["critical_fails"] = 0 # a success clears the streak
return True
except Exception as err: # best-effort: NEVER propagate
_debug(f"emit-exception {event_type}:{event_action}: {err!r}")
return False
def _cli(argv=None):
parser = argparse.ArgumentParser(
description="best-effort dd-orchestrator telemetry emitter (v1)")
parser.add_argument("event_type", choices=sorted(EVENT_TYPES))
parser.add_argument("--action", required=True, help="event lifecycle action")
parser.add_argument("--session-id", required=True,
help="the run's shared session id (from resolve.py 'SESSION ID:')")
parser.add_argument("--field", action="append", default=[], metavar="KEY=VALUE",
help="event field (repeatable); non-allow-listed keys are dropped")
args = parser.parse_args(argv)
fields = {}
for item in args.field:
if "=" not in item:
_debug(f"cli: ignoring malformed --field {item!r}")
continue
key, value = item.split("=", 1)
fields[key.strip()] = value
emit(args.event_type, args.action, args.session_id, fields)
return 0 # ALWAYS succeed — telemetry must never fail the caller
if __name__ == "__main__":
sys.exit(_cli())
Requirements and compatibility
Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta
“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.
Needs: Needs an API key, Needs Python, Needs Node.js
Security
What our scanner noticed· Low risk
- Scripts that your agent may run
This skill includes scripts that your agent may run. Read them before you use it.
A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.
Attribution and licence
- Published by
- Datadog
- Publisher
- Datadog (Official)
- Source
- datadog-labs/agent-skills · dd-orchestrator · commit 8e1306a
- Licence
- MIT License (redistribution allowed)
- LICENSE (added by AutoGineer from LICENSE)
- ATTRIBUTION.md (written by AutoGineer)
- Synced
AutoGineer is not affiliated with or endorsed by Datadog or Datadog.
Download ZIP
The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download dd-orchestrator.zip