diff --git a/scripts/l3b_agent_stub.py b/scripts/l3b_agent_stub.py new file mode 100755 index 0000000..7eda0b3 --- /dev/null +++ b/scripts/l3b_agent_stub.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""l3b_agent_stub.py — D-008 / D-026 / D-021 + +Parses a GitHub/Gitea Issue body by keywords and emits a contract.yaml that +selects an L2 stack. This is the agentic (L3B) entry surface: deterministic +keyword matching, no external AI APIs. + +D-008 keyword map (priority order — first match wins): + gas, price, ingest, data-lake -> l2-commodity-price-feed + invoice, billing -> l2-invoice-service + analytics, historical, query -> l2-energy-analytics-api + regulatory, compliance, reporting, trading + -> l2-regulatory-reporting + (no match) -> l2-invoice-service (fallback) + +Output contract.yaml (D-021 schema): + stack: + inputs: + environment: dev + owner: citizen-developer + source: l3b-agent-stub + public-ingress: false + +Input: + argv[1] = issue body text (or stdin if argv[1] absent/empty) + -o = write the contract to a file (default: stdout) +Exit: + 0 on success, 1 on empty input +""" +import sys + + +# Ordered keyword groups -> L2 stack mapping (D-008). First match wins. +KEYWORD_MAP = [ + (("gas", "price", "ingest", "data-lake"), "l2-commodity-price-feed"), + (("invoice", "billing"), "l2-invoice-service"), + (("analytics", "historical", "query"), "l2-energy-analytics-api"), + (("regulatory", "compliance", "reporting", "trading"), "l2-regulatory-reporting"), +] + +FALLBACK_STACK = "l2-invoice-service" + + +def map_issue_to_stack(text: str) -> str: + lowered = text.lower() + for keywords, stack in KEYWORD_MAP: + for kw in keywords: + if kw in lowered: + return stack + return FALLBACK_STACK + + +def render_contract(stack: str) -> str: + # Fixed-schema YAML (D-021). Emitted as text (no yaml dependency needed). + return ( + f"stack: {stack}\n" + "inputs:\n" + " environment: dev\n" + " owner: citizen-developer\n" + " source: l3b-agent-stub\n" + "public-ingress: false\n" + ) + + +def read_issue_body(argv: list) -> str: + """Read issue body from argv[1] or stdin. Empty -> error.""" + if len(argv) >= 2 and argv[1].strip(): + return argv[1] + # Fall back to stdin if argv[1] is absent or empty. + if not sys.stdin.isatty(): + data = sys.stdin.read() + if data.strip(): + return data + return "" + + +def parse_output_flag(argv: list): + """Extract -o from argv (returns (rest, output_path)).""" + output_path = None + rest = [] + i = 1 + while i < len(argv): + arg = argv[i] + if arg == "-o": + if i + 1 < len(argv): + output_path = argv[i + 1] + i += 2 + continue + else: + print("l3b_agent_stub: -o requires a path argument", file=sys.stderr) + sys.exit(1) + rest.append(arg) + i += 1 + return rest, output_path + + +def main() -> int: + rest, output_path = parse_output_flag(sys.argv) + body = read_issue_body(rest) + if not body.strip(): + print("l3b_agent_stub: empty issue body (no argv[1] and no stdin)", file=sys.stderr) + return 1 + + stack = map_issue_to_stack(body) + contract = render_contract(stack) + + if output_path: + with open(output_path, "w", encoding="utf-8") as fh: + fh.write(contract) + else: + sys.stdout.write(contract) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file