feat: inject project .dag as ground truth before markdown context

agent/prompt_builder.py

@@ -77,6 +77,65 @@ def _find_git_root(start: Path) -> Optional[Path]:
 _DAG_MD_NAMES = (".dag.md", "DAG.md")
 
 
+def _load_project_dag(cwd_path: Path) -> str:
+    """Inject project .dag files as DAG ground truth (before markdown).
+
+    Scans projects/ and specs/ subdirectories for .dag files.
+    Returns compact DAG-path notation: Entity-> Target:verb(card).
+    Max 3 project DAGs, each max 8,000 chars.
+    """
+    import json as _json
+    dag_files: List[Path] = []
+    for scan_dir in ("projects", "specs"):
+        scan_path = cwd_path / scan_dir
+        if scan_path.is_dir():
+            dag_files.extend(sorted(scan_path.rglob("*.dag")))
+    if not dag_files:
+        return ""
+
+    sections: List[str] = []
+    for dag_path in dag_files[:3]:
+        try:
+            content = dag_path.read_text(encoding="utf-8").strip()
+            if not content:
+                continue
+            dag = _json.loads(content)
+            project = dag.get("p", "?")
+            nodes = dag.get("n", [])
+            node_names: Dict[int, str] = {}
+            for n in nodes:
+                node_names[n[0]] = n[1]
+
+            lines = ["[%s]  # DAG ground truth" % project]
+            for n in nodes:
+                nid, name, ntype, _desc, _props, states, edges = n
+                # Compact edges: Target:verb(card)
+                edge_strs = []
+                for edge in edges:
+                    tid, verb, card = edge[0], edge[1], edge[2]
+                    tname = node_names.get(tid, "?")
+                    edge_strs.append("%s:%s(%s)" % (tname, verb, card))
+                if edge_strs:
+                    lines.append("%s-> %s" % (name, " | ".join(edge_strs)))
+                if states and len(states) > 0:
+                    lines.append("  states: %s" % " -> ".join(str(s) for s in states[:12]))
+
+            dag_text = "\n".join(lines)
+            try:
+                rel = str(dag_path.relative_to(cwd_path))
+            except ValueError:
+                rel = str(dag_path)
+            if len(dag_text) > 8000:
+                dag_text = dag_text[:8000] + "\n[...truncated]\n"
+            sections.append("[Subdirectory DAG context: %s]\n%s" % (rel, dag_text))
+        except Exception:
+            pass
+
+    if not sections:
+        return ""
+    return "\n\n".join(sections)
+
+
 def _find_dag_md(cwd: Path) -> Optional[Path]:
     """Discover the nearest ``.dag.md`` or ``DAG.md``.
 
@@ -1455,9 +1514,11 @@ def build_context_files_prompt(cwd: Optional[str] = None, skip_soul: bool = Fals
     cwd_path = Path(cwd).resolve()
     sections = []
 
-    # Priority-based project context: first match wins
+    # Priority-based project context: DAG files first (ground truth),
+    # then markdown context files (fallback).
     project_context = (
-        _load_dag_md(cwd_path)
+        _load_project_dag(cwd_path)
+        or _load_dag_md(cwd_path)
         or _load_agents_md(cwd_path)
         or _load_claude_md(cwd_path)
         or _load_cursorrules(cwd_path)

agent/subdirectory_hints.py

@@ -24,9 +24,10 @@
 logger = logging.getLogger(__name__)
 
 # Context files to look for in subdirectories, in priority order.
-# Same filenames as prompt_builder.py but we load ALL found (not first-wins)
-# since different subdirectories may use different conventions.
+# DAG files (.dag) are loaded FIRST — they provide ground-truth structure.
+# Markdown files are fallbacks for human-written prose context.
 _HINT_FILENAMES = [
+    "*.dag",           # DAG ground truth — machine-read structure
     "AGENTS.md", "agents.md",
     "CLAUDE.md", "claude.md",
     ".cursorrules",
@@ -195,6 +196,66 @@ def _is_valid_subdir(self, path: Path) -> bool:
                 return False
         return True
 
+    def _read_hint_file(self, hint_path: Path, filename: str, found_hints: list):
+        """Read a single hint file and append (rel_path, content) to found_hints.
+
+        For .dag files, converts JSON to compact DAG-path notation.
+        """
+        try:
+            content = hint_path.read_text(encoding="utf-8").strip()
+            if not content:
+                return
+            # Convert .dag JSON to compact DAG-path notation
+            if hint_path.suffix == ".dag":
+                content = self._compact_dag(content, hint_path)
+            # Security scan
+            content = _scan_context_content(content, filename)
+            if len(content) > _MAX_HINT_CHARS:
+                content = (
+                    content[:_MAX_HINT_CHARS]
+                    + f"\n\n[...truncated {filename}: {len(content):,} chars total]"
+                )
+            # Best-effort relative path for display
+            rel_path = str(hint_path)
+            try:
+                rel_path = str(hint_path.relative_to(self.working_dir))
+            except ValueError:
+                try:
+                    rel_path = str(hint_path.relative_to(Path.home()))
+                    rel_path = "~/" + rel_path
+                except ValueError:
+                    pass
+            found_hints.append((rel_path, content))
+        except Exception as exc:
+            logger.debug("Could not read %s: %s", hint_path, exc)
+
+    @staticmethod
+    def _compact_dag(content: str, path: Path) -> str:
+        """Convert .dag JSON to compact DAG-path notation for LLM consumption."""
+        import json
+        try:
+            dag = json.loads(content)
+            project = dag.get("p", path.stem)
+            nodes = dag.get("n", [])
+            node_names = {}
+            for n in nodes:
+                node_names[n[0]] = n[1]
+            lines = ["[%s]  # DAG ground truth" % project]
+            for n in nodes:
+                nid, name, ntype, _desc, _props, states, edges = n
+                edge_strs = []
+                for edge in edges:
+                    tid, verb, card = edge[0], edge[1], edge[2]
+                    tname = node_names.get(tid, "?")
+                    edge_strs.append("%s:%s(%s)" % (tname, verb, card))
+                if edge_strs:
+                    lines.append("%s-> %s" % (name, " | ".join(edge_strs)))
+                if states and len(states) > 0:
+                    lines.append("  states: %s" % " -> ".join(str(s) for s in states[:12]))
+            return "\n".join(lines)
+        except Exception:
+            return content  # fallback: raw content
+
     def _load_hints_for_directory(self, directory: Path) -> Optional[str]:
         """Load hint files from a directory. Returns formatted text or None.
 
@@ -220,38 +281,25 @@ def _load_hints_for_directory(self, directory: Path) -> Optional[str]:
 
         found_hints = []
         for filename in _HINT_FILENAMES:
-            hint_path = directory / filename
-            try:
-                if not hint_path.is_file():
-                    continue
-            except OSError:
-                continue
-            try:
-                content = hint_path.read_text(encoding="utf-8").strip()
-                if not content:
-                    continue
-                # Same security scan as startup context loading
-                content = _scan_context_content(content, filename)
-                if len(content) > _MAX_HINT_CHARS:
-                    content = (
-                        content[:_MAX_HINT_CHARS]
-                        + f"\n\n[...truncated {filename}: {len(content):,} chars total]"
-                    )
-                # Best-effort relative path for display
-                rel_path = str(hint_path)
+            # Handle glob patterns (e.g. "*.dag") — scan directory for matches
+            if "*" in filename or "?" in filename:
+                matches = sorted(directory.glob(filename))
+                for hint_path in matches:
+                    if not hint_path.is_file():
+                        continue
+                    self._read_hint_file(hint_path, filename, found_hints)
+                if found_hints:
+                    break  # first pattern that matches wins per directory
+            else:
+                hint_path = directory / filename
                 try:
-                    rel_path = str(hint_path.relative_to(self.working_dir))
-                except ValueError:
-                    try:
-                        rel_path = str(hint_path.relative_to(Path.home()))
-                        rel_path = "~/" + rel_path
-                    except ValueError:
-                        pass  # keep absolute
-                found_hints.append((rel_path, content))
-                # First match wins per directory (like startup loading)
-                break
-            except Exception as exc:
-                logger.debug("Could not read %s: %s", hint_path, exc)
+                    if not hint_path.is_file():
+                        continue
+                except OSError:
+                    continue
+                self._read_hint_file(hint_path, filename, found_hints)
+                if found_hints:
+                    break  # first match wins per directory
 
         if not found_hints:
             return None

skills/intelligence-engine/scripts/repo-audit

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""repo-audit → 5-line max. True secrets/emails only. Exits 0=clean 1=found."""
+import re, subprocess, sys
+from pathlib import Path
+
+REPO = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()
+SKIP = {'.git', '.venv', 'node_modules', '__pycache__', 'egg-info', '.tox', 'dist', 'build'}
+
+def walk():
+    for f in REPO.rglob("*"):
+        parts = set(f.parts)
+        if parts & SKIP or f.is_dir():
+            continue
+        if f.suffix in ('.py','.md','.toml','.yaml','.yml','.json','.cfg','.sh'):
+            try: yield f, f.read_text()
+            except: pass
+
+def audit():
+    findings = []
+    # Secret: only hardcoded literal values (not env var usage)
+    secret = re.compile(r'(?:api_key|secret|token|password|AUTH_TOKEN)\s*[:=]\s*["\']([^"\' ]{16,})["\']', re.I)
+    email_re = re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}')
+    ok_email = {'logohere@gmail.com','noreply@github.com','users.noreply.github.com',
+                'git@github.com','noreply@nousresearch.com','noreply@openai.com'}
+    fake = {'example','test','domain','foo','bar','user@','email@','name@','john@','alice@',
+            'noreply','no-reply','donotreply','placeholder','sample','fake','dummy'}
+
+    for path, text in walk():
+        # True hardcoded secrets
+        for m in secret.finditer(text):
+            val = m.group(1)
+            if val != '${' and 'os.getenv' not in text[m.start():m.start()+80]:
+                findings.append(f"SECRET: {m.group(0)[:60]} in {path.name}")
+
+        # Suspicious emails (not known-safe, not obvious test fixtures)
+        for m in email_re.finditer(text):
+            e = m.group().lower()
+            if e in ok_email or any(f in e for f in fake):
+                continue
+            if path.suffix in ('.py','.sh','.yaml','.yml','.toml','.cfg'):
+                findings.append(f"EMAIL: {e} in {path.name}")
+
+    # Untracked
+    s = subprocess.run(["git","status","--porcelain"], capture_output=True, text=True, cwd=REPO)
+    untracked = [l for l in s.stdout.splitlines() if l.startswith("??")]
+    if untracked:
+        findings.append(f"UNTRACKED: {len(untracked)} files")
+
+    # Version drift
+    try:
+        ppt = REPO / "pyproject.toml"
+        m = re.search(r'version\s*=\s*"([^"]+)"', ppt.read_text())
+        tags = subprocess.run(["git","tag","--sort=-v:refname"], capture_output=True, text=True, cwd=REPO)
+        latest = [t for t in tags.stdout.splitlines() if t.startswith('v')]
+        if m and latest and f"v{m.group(1)}" != latest[0]:
+            findings.append(f"TAG: pyproject={m.group(1)} latest={latest[0]}")
+    except: pass
+
+    return findings
+
+if __name__ == "__main__":
+    f = audit()
+    if f:
+        for line in f[:12]:
+            print(f"  ! {line}")
+        if len(f) > 12:
+            print(f"  … +{len(f)-12} more")
+        sys.exit(1)
+    else:
+        p = REPO / "pyproject.toml"
+        v = re.search(r'version\s*=\s*"([^"]+)"', p.read_text()).group(1) if p.exists() else "?"
+        t = subprocess.run(["git","describe","--tags","--always"], capture_output=True, text=True, cwd=REPO)
+        u = subprocess.run(["git","status","--porcelain"], capture_output=True, text=True, cwd=REPO)
+        n = sum(1 for l in u.stdout.splitlines() if l.startswith("??"))
+        print(f"  clean — v{v} ({t.stdout.strip()}) — {n} untracked — no secrets")