#!/usr/bin/env python3 """Ask → wait on subscribe → list and download session artifacts. Use when you want every downloadable note from a finished Ask (operator follow-through, product changelog, triage reports), not a single named file. Docs: https://appcd-dev.github.io/stackgen-sdk/code-samples/ask-session-artifacts/ Usage: export STACKGEN_URL=https://app.stackgen.com export STACKGEN_TOKEN=stackgen_… export STACKGEN_WORKSPACE_ID= python examples/ask/ask-session-artifacts.py \\ --message 'Look at recent org changes and leave me a markdown note I can download' # optional pin: # --agent my-ops-agent # --out-dir ./out/session-artifacts """ from __future__ import annotations import argparse import json import os import re import sys from pathlib import Path from stackgen import StackgenClient, StackgenConfig, TimeoutError as SdkTimeoutError # Keep a Changelog section titles (https://keepachangelog.com/). KEEP_A_CHANGELOG_SECTIONS = ( "Added", "Changed", "Deprecated", "Removed", "Fixed", "Security", ) def env(name: str, *fallbacks: str) -> str: for key in (name, *fallbacks): value = (os.environ.get(key) or "").strip() if value: return value raise SystemExit(f"missing env {name}") def looks_like_text(name: str, data: bytes) -> bool: lower = name.lower() if re.search(r"\.(md|txt|markdown|json)$", lower): return True sample = data[:512].decode("utf-8", errors="replace") return bool(sample) and "\0" not in sample def keep_a_changelog_hits(text: str) -> list[str]: hits: list[str] = [] for section in KEEP_A_CHANGELOG_SECTIONS: if re.search(rf"^#{{1,3}}\s*{re.escape(section)}\b", text, flags=re.M | re.I): hits.append(section) return hits def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--message", default=( "Look at recent changes in our GitHub org. Leave me a short " "markdown note I can download covering what you found." ), help="Natural-language Ask message", ) parser.add_argument("--agent", help="Pin an agent via entity_refs") parser.add_argument("--workflow-name", help="Pin a workflow via entity_refs") parser.add_argument( "--out-dir", default="session-artifacts", help="Directory for downloaded artifact files", ) parser.add_argument( "--timeout-ms", type=int, default=30 * 60 * 1000, help="Subscribe wait budget in milliseconds", ) args = parser.parse_args() client = StackgenClient( StackgenConfig( base_url=env("STACKGEN_URL"), api_token=env("STACKGEN_TOKEN"), workspace_id=env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"), timeout_seconds=max(30, args.timeout_ms // 1000), ) ) entity_refs: list[dict[str, str]] = [] if args.agent: entity_refs.append({"name": args.agent, "type": "agent"}) if args.workflow_name: entity_refs.append({"name": args.workflow_name, "type": "workflow"}) session_id = "" try: start = client.aiden.start_ask( args.message, entity_refs=entity_refs or None, ) session_id = str(start.get("session_id") or "") trace_id = str(start.get("trace_id") or "") if not session_id or not trace_id: raise SystemExit(f"start_ask missing session_id/trace_id: {start}") print(f"started session_id={session_id} trace_id={trace_id}", file=sys.stderr) try: done = client.aiden.wait_for_subscribe_complete( trace_id, timeout_ms=args.timeout_ms, ) if not done.ok: print( f"subscribe incomplete ({done}); waiting on execution status", file=sys.stderr, ) trace = client.aiden.wait_for_execution(trace_id) status = str((trace.get("execution") or {}).get("status") or "") print(f"execution status={status}", file=sys.stderr) else: print(f"subscribe complete ok={done.ok}", file=sys.stderr) except SdkTimeoutError as exc: print( f"subscribe timed out ({exc}); waiting on execution status", file=sys.stderr, ) trace = client.aiden.wait_for_execution(trace_id) status = str((trace.get("execution") or {}).get("status") or "") print(f"execution status={status}", file=sys.stderr) out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) downloaded = client.aiden.download_session_artifacts(session_id, str(out_dir)) for file in downloaded: print( f"downloaded {file.get('name')} -> {file.get('path')}", file=sys.stderr, ) if not downloaded: export = client.aiden.export_session(session_id, parts="report") parts = export.get("parts") or [] report = next( ( str(part.get("content") or "") for part in parts if isinstance(part, dict) and part.get("kind") == "report" ), "", ) if not report.strip(): raise SystemExit( f"no session artifacts and no export report for session_id={session_id}" ) dest = out_dir / "session-report.md" dest.write_text(report, encoding="utf-8") downloaded.append( {"name": "session-report.md", "path": str(dest), "meta": {}} ) print(f"wrote export report -> {dest}", file=sys.stderr) text_notes: list[str] = [] for file in downloaded: data = Path(file["path"]).read_bytes() if not looks_like_text(file["name"], data): continue text = data.decode("utf-8", errors="replace").strip() if not text: continue text_notes.append(text) sections = keep_a_changelog_hits(text) if sections: print( f"{file['name']}: Keep a Changelog sections found: " + ", ".join(sections), file=sys.stderr, ) if text_notes: combined = out_dir / "combined-note.md" combined.write_text("\n\n---\n\n".join(text_notes), encoding="utf-8") print(f"wrote {combined}", file=sys.stderr) blob = "\n".join(text_notes) urls = re.findall(r"https?://[^\s)\]>]+", blob) console_urls = [ u for u in urls if re.search(r"grafana|console_url|/d/", u, flags=re.I) ][:5] print( json.dumps( { "session_id": session_id, "trace_id": trace_id, "artifacts": downloaded, "console_urls": console_urls, }, indent=2, ) ) return 0 finally: if session_id: try: client.aiden.terminate_session(session_id, reason="sdk example finished") except Exception as exc: # noqa: BLE001 print(f"terminate session {session_id}: {exc}", file=sys.stderr) if __name__ == "__main__": raise SystemExit(main())