feat(P03): l3b_agent_stub.py (T-3.9)

---ci--- phase: 3 milestone: v1.0 status: execute persona: backend-engineer task: T-3.9 requirements: covered: [REQ-12] ---/ci--- Wave 2, task T-3.9. L3B keyword parser: maps issue body text to L2 stack (D-008) and emits contract.yaml (D-021).
2026-07-21 13:26:20 +00:00
parent 28535f9f5f
commit 464131fdf4
1 changed files with 117 additions and 0 deletions
@@ -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: <mapped L2 name>
+  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 <path> = 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 <path> 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())