fix: stabilize session history and recovery, add Claude quota refresh

cc_remote/claude_service/client.py

@@ -23,6 +23,24 @@
     "claude_service_callback", default=None)
 
 
+def service_owner_exited(identity) -> bool:
+    """Prove an exact local owner exited; an unreadable PID is not proof."""
+    if identity is None:
+        return False
+    from cc_remote.wrapper.process_scan import process_identity
+
+    current = process_identity(identity.pid)
+    if current is not None:
+        return current != identity
+    try:
+        os.kill(identity.pid, 0)
+    except ProcessLookupError:
+        return True
+    except OSError:
+        pass
+    return False
+
+
 class Connection:
     def __init__(self, socket_path: str):
         self.socket_path = os.path.expanduser(socket_path)
@@ -101,7 +119,8 @@ def options_payload(options) -> dict:
 
 
 class RemoteClient:
-    def __init__(self, socket_path, *, options, metadata, isolated=False):
+    def __init__(self, socket_path, *, options, metadata, isolated=False,
+                 previous_owner_identity=None):
         self.connection = Connection(socket_path)
         self.options = options
         self.metadata = metadata
@@ -119,6 +138,7 @@ def __init__(self, socket_path, *, options, metadata, isolated=False):
         self.last_seq = 0
         self.ready = asyncio.Event()
         self.owner_identity = None
+        self.previous_owner_identity = previous_owner_identity
 
     async def connect(self) -> None:
         await self.connection.connect()
@@ -136,12 +156,32 @@ async def connect(self) -> None:
                 claude_sdk_process_env(self.options.env, environment)
                 if self.isolated else {**environment, **self.options.env}
             )
+            worker_id = self.metadata.get("service_id")
+            from cc_remote.wrapper.process_scan import process_identity
+
+            owner = process_identity(hello["pid"])
+            if (worker_id and self.options.resume and not self.options.fork_session
+                    and owner is not None and owner != self.previous_owner_identity
+                    and service_owner_exited(self.previous_owner_identity)):
+                sessions = await self.connection.call("list")
+                if not any(item["id"] == worker_id or all(
+                    item["metadata"].get(key) == self.metadata.get(key)
+                    for key in ("profile_root", "session_id", "space", "work_id", "btw")
+                ) for item in sessions):
+                    # A restarted service has no old in-memory worker. Resume
+                    # the same native transcript using its full account/cwd
+                    # identity, without submitting any prior accepted input.
+                    # Keep the hint until open succeeds so a failed connection
+                    # does not erase recovery authority. A replacement worker
+                    # already owning this transcript needs ordinary replay
+                    # recovery, not this idle control/new-input reconnect.
+                    worker_id = None
             self.description = await self._open({
                 "options": payload,
                 "metadata": self.metadata,
                 "isolated": self.isolated,
                 "fork": self.options.fork_session,
-                "session": self.metadata.get("service_id"),
+                "session": worker_id,
                 "strict_session": bool(hello.get("strict_controller_leases")),
             }, legacy=not hello.get("strict_controller_leases"))
         except BaseException:

cc_remote/claude_service/server.py

@@ -19,7 +19,9 @@
 from pathlib import Path
 from uuid import uuid4
 
-from cc_remote.claude_steering import PendingSteers, background_end_ids, is_managed_input, steer_message
+from cc_remote.claude_steering import (
+    PendingSteers, background_end_ids, is_human_result, is_managed_input, steer_message,
+)
 
 from .wire import (
     ControllerLeaseConflict, decode_sdk, encode_sdk, private_directory,
@@ -79,9 +81,7 @@ def close(self) -> None:
 
 
 def _human_result(data: dict) -> bool:
-    origin = data.get("origin")
-    kind = origin.get("kind") if isinstance(origin, dict) else None
-    return data.get("type") == "result" and kind in (None, "human")
+    return is_human_result(data)
 
 
 class Session:
@@ -282,6 +282,7 @@ def description(self) -> dict:
             "native_steering": True,
             "background_steering": True,
             "background_activity_steering": True,
+            "human_result_receipts": True,
             "pending_steers": {uid: {"id": data["id"]} for uid, data in self.steers.pending.items()},
         }
 
@@ -407,6 +408,7 @@ async def stream():
                 self.turn = {**params["turn"], "start_seq": self.journal.seq, "started_at": time.time(),
                              "previous_origin_id": self.origin_id}
                 self.managed_input_seen = False
+                self.steers.begin_turn()
                 self.pending_compact = (isinstance(prompt, str)
                                         and prompt.split(maxsplit=1)[:1] == ["/compact"])
                 self.terminal_seq = None

cc_remote/claude_steering.py

@@ -68,6 +68,17 @@ def is_managed_input(value: dict, *, pending_compact: bool = False) -> bool:
     return False
 
 
+def is_human_result(value: dict) -> bool:
+    """Use an exact consumption receipt before falling back to legacy origin."""
+    if (value.get("type") != "result" or value.get("parent_tool_use_id")
+            or value.get("parentToolUseID")):
+        return False
+    if "__cc_managed_result" in value:
+        return value["__cc_managed_result"] is True
+    origin = value.get("origin")
+    return not isinstance(origin, dict) or origin.get("kind") in (None, "human")
+
+
 class PendingSteers:
     """Fence accepted inputs against their exact replayed human UUIDs.
 
@@ -83,6 +94,10 @@ def __init__(self):
         self.background_origin: str | None = None
         self.background_origin_data: dict | None = None
         self._backgrounds: dict[str, dict] = {}
+        self._consumed_user_id: str | None = None
+
+    def begin_turn(self) -> None:
+        self._consumed_user_id = None
 
     def handoff_background(self, identity: str | None) -> None:
         self._backgrounds.pop(identity, None)
@@ -96,10 +111,31 @@ def add(self, native_id: str, metadata: dict) -> None:
             raise ClaudeSteerRejected("Claude steering capacity reached")
         self.pending[native_id] = metadata
 
-    def annotate(self, value: dict, *, managed_active: bool = False) -> dict:
+    def annotate(self, value: dict, *, managed_active: bool = False,
+                 result_receipts: bool = True) -> dict:
         origin = value.get("origin")
         kind = origin.get("kind") if isinstance(origin, dict) else None
         child = value.get("parent_tool_use_id") or value.get("parentToolUseID")
+        uid = value.get("uuid")
+        if (is_managed_input(value) and value.get("type") == "user"
+                and isinstance(uid, str) and uid
+                and (managed_active or uid in self.pending or value.get("__cc_steer"))):
+            self._consumed_user_id = uid
+        if value.get("type") == "result" and not child:
+            receipts = value.get("user_message_uuids")
+            receipts = receipts if isinstance(receipts, list) else []
+            receipts = [uid for uid in receipts if isinstance(uid, str) and uid]
+            receipt = value.get("user_message_uuid")
+            if isinstance(receipt, str) and receipt:
+                receipts.append(receipt)
+            if receipts and result_receipts and "__cc_managed_result" not in value:
+                # Task notifications can be absorbed into the human response.
+                # The Result must consume the latest echoed input, not merely
+                # an older input in the same session. Preserve this decision in
+                # the journal: the pinned SDK parser drops the receipt fields.
+                value = {**value, "__cc_managed_result":
+                         self._consumed_user_id is not None
+                         and self._consumed_user_id in receipts}
         if not child:
             if managed_active and is_managed_input(value, pending_compact=True):
                 # A native request can precede its replayed human input. That
@@ -150,7 +186,7 @@ def annotate(self, value: dict, *, managed_active: bool = False) -> dict:
             elif value.get("type") == "result":
                 ended = background_end_ids(value)
                 local_ends = ()
-                if kind in (None, "human"):
+                if is_human_result(value):
                     # An unattributed Result closes the physical response,
                     # including its in-turn task inputs. An older autonomous
                     # response must not be consumed by a new human terminal.
@@ -159,7 +195,7 @@ def annotate(self, value: dict, *, managed_active: bool = False) -> dict:
                                            if entry["managed"])
                     elif kind is None:
                         local_ends = tuple(self._backgrounds)
-                else:
+                elif isinstance(origin, dict) and kind not in (None, "human"):
                     # A precise unrelated origin remains authoritative.
                     local_ends = tuple(key for key, entry in self._backgrounds.items()
                                        if entry["origin_key"] in (None, _origin_key(origin)))
@@ -184,14 +220,16 @@ def annotate(self, value: dict, *, managed_active: bool = False) -> dict:
             if metadata is not None:
                 return {**value, "type": "system", "subtype": "cc_remote_steer_cancelled",
                         "__cc_steer_cancelled": metadata}
+        if value.get("type") == "result" and is_human_result(value):
+            self._consumed_user_id = None
+            if self.pending:
+                return {**value, "__cc_steer_intermediate": True}
         if kind not in (None, "human") or child:
             return value
         if value.get("type") == "user":
             metadata = self.pending.pop(value.get("uuid"), None)
             if metadata is not None:
                 return {**value, "__cc_steer": value.get("__cc_steer", metadata)}
-        elif value.get("type") == "result" and self.pending:
-            return {**value, "__cc_steer_intermediate": True}
         return value
 
     async def interrupt(self, client) -> None: