Add annotate for bulk comments/renames, and say what to do about --raw

declib/api/decompiler_client.py

@@ -595,6 +595,13 @@ def decompile_many(self, addrs: List[int], **kwargs) -> Dict[int, Optional[str]]
             "args": [list(addrs)], "kwargs": kwargs,
         })
 
+    def apply_annotations(self, items: List[Dict]) -> Dict[str, int]:
+        """Apply many comments/renames in one round-trip."""
+        return self._send_request({
+            "type": "method_call", "method_name": "apply_annotations",
+            "args": [list(items)],
+        })
+
     def disassemble_range(self, start: int, end: int, **kwargs) -> Optional[str]:
         """Disassemble an arbitrary address span, function or not."""
         return self._send_request({

declib/api/decompiler_interface.py

@@ -409,6 +409,80 @@ def decompile_many(self, addrs: List[int], **kwargs) -> Dict[int, Optional[str]]
                 dec = None
             out[addr] = dec.text if dec is not None else None
         return out
+    def apply_annotations(self, items: List[Dict]) -> Dict[str, int]:
+        """Apply many comments and/or function renames in one call.
+
+        Each item is ``{"addr": int, "comment": str?, "name": str?}``. Both
+        fields are optional; an item may carry either or both.
+
+        Like :meth:`decompile_many`, the default loops in-process — which
+        through DecompilerServer collapses N round-trips into one. Annotation
+        is naturally produced in bulk (a dict of address to label, built while
+        reading a binary), and applying it one CLI call at a time is what
+        pushes people back to writing a script against the raw backend API.
+
+        A failed item is counted, not raised, so one bad address does not
+        discard the rest of the batch.
+
+        @param items: annotation records, as above.
+        @return: {"comments": n, "names": n, "failed": n}
+        """
+        applied = {"comments": 0, "names": 0, "failed": 0, "errors": []}
+
+        def _fail(addr, what, reason):
+            # str(exc) is empty for exceptions raised without arguments, which
+            # would surface as "failed 0x...: " with no reason at all. Name the
+            # type in that case rather than reporting a blank.
+            text = str(reason)
+            if not text.strip() and isinstance(reason, BaseException):
+                text = f"{type(reason).__name__} (no message)"
+            applied["failed"] += 1
+            applied["errors"].append({
+                "addr": addr, "field": what, "error": text,
+            })
+
+        for item in items:
+            addr = item.get("addr")
+            if addr is None:
+                _fail(None, "addr", "record has no address")
+                continue
+
+            comment = item.get("comment")
+            name = item.get("name")
+            if comment is None and name is None:
+                _fail(addr, "record", "neither 'comment' nor 'name' given")
+                continue
+
+            if comment is not None:
+                try:
+                    self.comments[addr] = Comment(addr=addr, comment=comment)
+                    applied["comments"] += 1
+                except Exception as e:  # noqa: BLE001
+                    self.warning(f"apply_annotations: comment at {hex(addr)} failed: {e}")
+                    _fail(addr, "comment", e)
+
+            if name is not None:
+                try:
+                    func = self.functions[addr]
+                    if func is None:
+                        raise ValueError("no function starts at this address")
+                    func.name = name
+                    if func.header is not None:
+                        func.header.name = name
+                    self.functions[addr] = func
+                    # Read back rather than assuming: a backend can accept the
+                    # write and quietly not apply it, and a silently-miscounted
+                    # rename is worse than a reported failure.
+                    check = self.functions[addr]
+                    if check is not None and getattr(check, "name", None) != name:
+                        raise ValueError(
+                            f"backend kept the name {getattr(check, 'name', None)!r}"
+                        )
+                    applied["names"] += 1
+                except Exception as e:  # noqa: BLE001
+                    self.warning(f"apply_annotations: rename at {hex(addr)} failed: {e}")
+                    _fail(addr, "name", e)
+        return applied
 
     def xrefs_to(self, artifact: Artifact, decompile=False, only_code=False) -> List[Artifact]:
         """

declib/cli/decompiler_cli.py

@@ -797,10 +797,20 @@ def _execute_batch_operation(operation: Dict, batch_args) -> Dict:
             )
     for option in ("--raw", "--help", "-h"):
         if _batch_option_present(argv, option):
+            # Rejecting --raw is correct: batch results are structured, and raw
+            # text printed straight to stdout would corrupt them. But saying
+            # only "not allowed" leaves the caller guessing, so name the
+            # replacement -- the text they wanted is already in the result.
+            if option == "--raw":
+                error = ("--raw is not allowed inside a structured batch: batch "
+                         "results are always JSON. Drop it — each entry's text "
+                         "is in results[].result.text.")
+            else:
+                error = f"{option} is not allowed inside a structured batch."
             return _batch_result(
                 operation,
                 exit_code=EXIT_USER_ERROR,
-                error=f"{option} is not allowed inside a structured batch.",
+                error=error,
                 duration_ms=0,
             )
 
@@ -2145,6 +2155,76 @@ def cmd_export(args) -> int:
                 print(f"  {len(failed)} function(s) did not decompile "
                       f"(listed in export.json)")
     return EXIT_OK
+def cmd_annotate(args) -> int:
+    """Apply a batch of comments and renames in one call.
+
+    Annotation is produced in bulk -- you read a binary, build up a mapping of
+    address to label, and want to write it back. Doing that one `comment set`
+    at a time is N round-trips, which is why people reach for the raw backend
+    API instead.
+
+    Input is JSON on stdin (or --file): either a list of records, or an object
+    mapping address to a comment. Records look like:
+
+        {"addr": "0x401000", "comment": "parses the header", "name": "parse_hdr"}
+    """
+    raw = Path(args.file).read_text() if args.file else sys.stdin.read()
+    if not raw.strip():
+        raise SystemExit("no annotation input (give JSON on stdin or --file)")
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as e:
+        raise SystemExit(f"input is not valid JSON: {e}")
+
+    # Accept the shorthand {"0x401000": "a comment", ...} as well as records,
+    # because that is the shape people already build by hand.
+    records: List[Dict] = []
+    if isinstance(parsed, dict):
+        for key, value in parsed.items():
+            if isinstance(value, dict):
+                records.append({"addr": key, **value})
+            else:
+                records.append({"addr": key, "comment": value})
+    elif isinstance(parsed, list):
+        records = list(parsed)
+    else:
+        raise SystemExit("expected a JSON list of records or an object keyed by address")
+
+    with _with_client(args) as client:
+        items: List[Dict] = []
+        for rec in records:
+            if not isinstance(rec, dict) or "addr" not in rec:
+                raise SystemExit(f"every record needs an 'addr': {rec!r}")
+            addr_value, _ = _parse_target(str(rec["addr"]))
+            if addr_value is None:
+                raise SystemExit(f"invalid address {rec['addr']!r}")
+            item = {"addr": _to_lifted_addr(client, addr_value)}
+            if rec.get("comment") is not None:
+                item["comment"] = str(rec["comment"])
+            if rec.get("name") is not None:
+                item["name"] = str(rec["name"])
+            if len(item) == 1:
+                raise SystemExit(
+                    f"record for {rec['addr']} has neither 'comment' nor 'name'"
+                )
+            items.append(item)
+
+        if not items:
+            raise SystemExit("nothing to apply")
+
+        result = client.apply_annotations(items) or {}
+        result["requested"] = len(items)
+        if args.json:
+            _emit(args, result)
+        else:
+            print(f"applied {result.get('comments', 0)} comment(s) and "
+                  f"{result.get('names', 0)} rename(s) from {len(items)} record(s)")
+            for err in result.get("errors") or []:
+                where = err.get("addr")
+                where = _format_addr_hex(where) if isinstance(where, int) else where
+                print(f"  failed {where} ({err.get('field')}): {err.get('error')}",
+                      file=sys.stderr)
+    return EXIT_OK if not result.get("failed") else EXIT_USER_ERROR
 
 
 def cmd_get_callers(args) -> int:
@@ -3097,6 +3177,15 @@ def build_parser() -> argparse.ArgumentParser:
     _add_server_filter_args(p_export)
     _add_output_args(p_export)
     p_export.set_defaults(func=cmd_export)
+    p_annotate = sub.add_parser(
+        "annotate",
+        help="Apply many comments/renames at once from JSON (stdin or --file).",
+    )
+    p_annotate.add_argument("--file",
+                            help="Read JSON from this path instead of stdin.")
+    _add_server_filter_args(p_annotate)
+    _add_output_args(p_annotate)
+    p_annotate.set_defaults(func=cmd_annotate)
 
     p_lf = sub.add_parser("list_functions", help="List functions in the binary.")
     p_lf.add_argument("--filter", dest="filter", help="Regex to filter function names.")

tests/test_decompiler_cli.py

@@ -2250,6 +2250,76 @@ def test_safe_filename_is_filesystem_safe_and_unique(self):
         self.assertEqual(_safe_filename("", 0x55), "00000055.c")
         # Two same-named functions at different addresses cannot collide.
         self.assertNotEqual(_safe_filename("f", 1), _safe_filename("f", 2))
+class TestAnnotateBatch(unittest.TestCase):
+    """Bulk annotation plumbing that needs no backend."""
+
+    def _apply(self, items):
+        """Run apply_annotations against a stand-in interface."""
+        from declib.api.decompiler_interface import DecompilerInterface
+
+        class _Fake(DecompilerInterface):
+            def __init__(self):
+                self.comments = {}
+                self.functions = {}
+                self._warnings = []
+
+            def warning(self, msg):
+                self._warnings.append(msg)
+
+        fake = _Fake()
+        return fake, DecompilerInterface.apply_annotations(fake, items)
+
+    def test_missing_address_is_reported_not_raised(self):
+        _, res = self._apply([{"comment": "orphan"}])
+        self.assertEqual(res["failed"], 1)
+        self.assertEqual(res["errors"][0]["field"], "addr")
+
+    def test_record_with_neither_field_is_reported(self):
+        _, res = self._apply([{"addr": 0x1000}])
+        self.assertEqual(res["failed"], 1)
+        self.assertIn("neither", res["errors"][0]["error"])
+
+    def test_one_bad_record_does_not_discard_the_batch(self):
+        fake, res = self._apply([
+            {"addr": 0x1000, "comment": "kept"},
+            {"addr": 0x2000},                      # invalid
+            {"addr": 0x3000, "comment": "also kept"},
+        ])
+        self.assertEqual(res["comments"], 2)
+        self.assertEqual(res["failed"], 1)
+
+    def test_failure_reason_is_never_blank(self):
+        """A bare exception must not surface as 'failed 0x...: '."""
+        from declib.api.decompiler_interface import DecompilerInterface
+
+        class _Exploding(DecompilerInterface):
+            def __init__(self):
+                self.comments = self
+                self.functions = {}
+
+            def __setitem__(self, key, value):
+                raise KeyError()      # empty str(), the blank-error case
+
+            def warning(self, msg):
+                pass
+
+        fake = _Exploding()
+        res = DecompilerInterface.apply_annotations(fake, [{"addr": 0x1000, "comment": "x"}])
+        self.assertEqual(res["failed"], 1)
+        text = res["errors"][0]["error"]
+        self.assertTrue(text.strip(), "error text was blank")
+        self.assertIn("KeyError", text)
+
+
+class TestBatchRawMessage(unittest.TestCase):
+    def test_raw_rejection_names_the_replacement(self):
+        """Rejecting --raw is right; leaving the caller guessing is not."""
+        from declib.cli.decompiler_cli import _execute_batch_operation
+
+        result = _execute_batch_operation(
+            {"id": "x", "argv": ["decompile", "0x1000", "--raw"]}, None
+        )
+        self.assertIn("results[].result.text", result["error"])
 class TestServerStartupDiagnostics(unittest.TestCase):
     """Startup/teardown diagnostics. No backend required."""