Trigger Policy

Contract

Iron law: merge and release MUST appear in every trigger's policy.forbidden_actions — triggers cannot grant Tier 3 autonomy regardless of any other configuration. Invoked in validate mode (read-only) by /flow:trigger create|enable|validate and by /flow:watch step 4 before a trigger YAML is written or enabled, and in enforce mode by /flow:run step 3 and /flow:trigger run before the target command is dispatched. Returns the JSON report below with exit 0 (pass or soft warning), 1 (policy violation), or 2 (schema invalid); the caller aborts on non-zero. Permitted skips: none — every step runs; the only non-blocking outcome is concurrency_warning.

Inputs

The invoking command MUST pass:

1. Trigger YAML path — a template under plugins/flow/triggers/templates/ or a project-local .flow/triggers/<id>.trigger.yaml.
2. Mode — validate | enforce.

Output

{
  "trigger_id": "pr-123-watch",
  "schema_valid": true,
  "tier3_violations": [],
  "recursion_violations": [],
  "missing_required_forbidden": [],
  "cross_reference_violations": [],
  "concurrency_violations": [],
  "overall": "pass"
}

Steps

1. Schema validation: load the YAML and validate it against the trigger schema. Exit 2 → overall: schema_invalid.

   # PYTHONPATH and sys.path are cleaned first: the trigger comes with the
   # repository, and so could a planted jsonschema.py or sitecustomize.py.
   [ -n "${FLOW_USER_PYTHONPATH+x}" ] || export FLOW_USER_PYTHONPATH="${PYTHONPATH-}"
   _flow_pp=""; if [ -n "${PYTHONPATH-}" ]; then _flow_pp=$(python3 -I -c 'exec("import os, sys\ndef ids(p):\n    out = set()\n    while True:\n        try:\n            st = os.stat(p)\n        except OSError:\n            return out\n        out.add((st.st_dev, st.st_ino))\n        q = os.path.dirname(p)\n        if q == p:\n            return out\n        p = q\ntry:\n    cwd = os.getcwd()\nexcept OSError:\n    sys.exit(0)\ntop = d = cwd\nwhile True:\n    if os.path.lexists(os.path.join(d, \".git\")):\n        top = d\n        break\n    q = os.path.dirname(d)\n    if q == d:\n        break\n    d = q\nst = os.stat(top)\ntop_id = (st.st_dev, st.st_ino)\nup = ids(cwd)\nkeep = []\nfor e in os.environ.get(\"PYTHONPATH\", \"\").split(\":\"):\n    if not e.startswith(\"/\"):\n        continue\n    r = os.path.realpath(e)\n    if \":\" in r or chr(10) in r or not os.path.isdir(r):\n        continue\n    try:\n        st = os.stat(r)\n    except OSError:\n        continue\n    if (st.st_dev, st.st_ino) in up or top_id in ids(r):\n        continue\n    keep.append(r)\nsys.stdout.buffer.write(os.fsencode(\":\".join(keep)))")' 2>/dev/null) || _flow_pp=""; fi
   if [ -n "$_flow_pp" ]; then export PYTHONPATH="$_flow_pp"; else unset PYTHONPATH; fi
   python3 - "${TRIGGER_YAML}" "plugins/flow/schemas/v1/trigger.schema.json" <<'PYEOF'
   import os, sys
   try:
       _flow_cwd = os.path.realpath(os.getcwd())
   except OSError:
       _flow_cwd = None
   sys.path[:] = [p for p in sys.path if p and os.path.isabs(p) and os.path.realpath(p) != _flow_cwd]
   import json, yaml, jsonschema
   with open(sys.argv[1], encoding="utf-8") as f:
       doc = yaml.safe_load(f)
   with open(sys.argv[2], encoding="utf-8") as f:
       schema = json.load(f)
   validator = jsonschema.validators.validator_for(schema)(schema)
   errors = sorted(validator.iter_errors(doc), key=lambda e: list(e.path))
   for e in errors:
       print("SCHEMA_ERROR=%s: %s" % ("/".join(map(str, e.path)) or "(root)", e.message))
   sys.exit(2 if errors else 0)
   PYEOF

1. Tier 3 absolute deny: policy.forbidden_actions must contain both merge AND release. Missing either → tier3_violations.append({"action": "merge_or_release", "reason": "must be forbidden"}). Hard fail.
2. Recursion policy: recursion_policy.triggered_runs_may_create_triggers, triggered_runs_may_modify_triggers, and triggered_runs_may_enable_triggers must be false or unset (default false). Any true → recursion_violations; enabling it requires explicit Tier 3 authorization via AskUserQuestion at /flow:trigger create time.
3. Allowed types: trigger.type must be in flow.triggers.allowedTypes (cascade-resolved; default [manual, hook, loop_prompt]). github_actions | local_cron | local_daemon are schema-valid but not enabled by default — flow.triggers.allowedTypes must opt them in; surface as tier3_violations unless the project's setting permits them.
4. Active-trigger count: count .flow/triggers/*.trigger.yaml with metadata.enabled: true and lifecycle != disabled. If count >= flow.triggers.maxActiveTriggers (cascade-resolved; default 5), refuse to enable a new trigger — the user must /flow:trigger disable one first.
5. Concurrency sanity: concurrency.policy: cancel_previous on a trigger whose target invokes a Tier 2 action (push, commit) → concurrency_violations warning; cancel_previous + Tier 2 can produce partial commits. Soft warning only — never a hard fail.
6. Target workflow cross-reference: when target.workflow is set, the workflow must exist at plugins/flow/workflows/${WF}.workflow.yaml or .flow/workflows/${WF}.workflow.yaml ([ -f "$PLUGIN_PATH" ] || [ -f "$LOCAL_PATH" ]). Missing → cross_reference_violations.append({"type": "missing_target_workflow", "name": target_workflow, "checked_paths": [PLUGIN_PATH, LOCAL_PATH]}). Hard fail (exit 1) — a trigger pointing at a non-existent workflow is broken by construction; this catches typos (address vs address-pr) at creation rather than when /flow:run trigger <id> dispatches. When target.workflow is absent (target.command only), this step is a no-op.
7. Overall verdict:

Reuse map

• plugins/flow/schemas/v1/trigger.schema.json — schema validated against.
• plugins/flow/triggers/templates/ — plugin-shipped templates.
• plugins/flow/commands/trigger.md, watch.md, run.md — invoking commands.
• plugins/flow/references/flow-triggers.md — user-facing trigger documentation.