fix: bind client job events to command leases

src/archive_clients/state.py

@@ -38,6 +38,7 @@ class FileOperationConflict(RuntimeError):
 class CommandAcceptance:
     duplicate: bool
     acknowledgement_json: str
+    state: str
 
 
 class ClientStore:
@@ -85,7 +86,7 @@ class ClientStore:
             connection.execute("BEGIN IMMEDIATE")
             existing = connection.execute(
                 """
-                SELECT payload_sha256, command_json, acknowledgement_json
+                SELECT payload_sha256, command_json, acknowledgement_json, state
                 FROM commands WHERE command_id = ?
                 """,
                 (command_id,),
@@ -93,17 +94,26 @@ class ClientStore:
             if existing:
                 if existing["payload_sha256"] != digest or existing["command_json"] != payload:
                     raise CommandConflict("command ID was reused with different content")
-                return CommandAcceptance(True, existing["acknowledgement_json"])
+                return CommandAcceptance(
+                    True, existing["acknowledgement_json"], existing["state"]
+                )
+            status = json.loads(acknowledgement).get("status")
+            state = (
+                "rejected"
+                if status is not None
+                and status != "COMMAND_ACK_STATUS_ACCEPTED"
+                else "accepted"
+            )
             connection.execute(
                 """
                 INSERT INTO commands (
                     command_id, payload_sha256, command_json,
                     acknowledgement_json, state
-                ) VALUES (?, ?, ?, ?, 'received')
+                ) VALUES (?, ?, ?, ?, ?)
                 """,
-                (command_id, digest, payload, acknowledgement),
+                (command_id, digest, payload, acknowledgement, state),
             )
-        return CommandAcceptance(False, acknowledgement)
+        return CommandAcceptance(False, acknowledgement, state)
 
     def list_active_job_cursors(self) -> list[dict[str, object]]:
         with self._connect() as connection:
@@ -122,7 +132,7 @@ class ClientStore:
             rows = connection.execute(
                 """
                 SELECT command_id, command_json, acknowledgement_json
-                FROM commands ORDER BY rowid
+                FROM commands WHERE state = 'accepted' ORDER BY rowid
                 """
             ).fetchall()
         return [dict(row) for row in rows]
@@ -199,6 +209,91 @@ class ClientStore:
                 ),
             )
 
+    def ensure_job_definition(self, job_id: str, definition_json: str) -> None:
+        """Durably record an assignment without inventing a global event."""
+        definition = _canonical(json.loads(definition_json))
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            existing = connection.execute(
+                "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,)
+            ).fetchone()
+            if existing is not None:
+                if existing["definition_json"] != definition:
+                    raise JobConflict("job definition is immutable")
+                return
+            connection.execute(
+                """
+                INSERT INTO jobs (job_id, definition_json, state, revision,
+                                  last_event_sequence, committed)
+                VALUES (?, ?, 'JOB_STATE_QUEUED', 0, 0, 0)
+                """,
+                (job_id, definition),
+            )
+
+    def is_command_active(self, command_id: str) -> bool:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT state FROM commands WHERE command_id = ?", (command_id,)
+            ).fetchone()
+        return row is not None and row["state"] == "accepted"
+
+    def reconcile_job(
+        self,
+        *,
+        job_id: str,
+        definition_json: str,
+        state: str,
+        revision: int,
+        last_event_sequence: int,
+        committed: bool,
+        superseded_command_ids: list[str],
+    ) -> None:
+        """Apply control's cursor, retiring only stale command leases.
+
+        This drops unacknowledged local journal rows beyond control's cursor;
+        resource files and operation artifacts are intentionally retained.
+        """
+        definition = _canonical(json.loads(definition_json))
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            existing = connection.execute(
+                "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,)
+            ).fetchone()
+            if existing is not None and existing["definition_json"] != definition:
+                raise JobConflict("reconciliation has a different job definition")
+            connection.execute(
+                "DELETE FROM events WHERE job_id = ? AND sequence > ?",
+                (job_id, last_event_sequence),
+            )
+            connection.execute(
+                """
+                INSERT INTO jobs (job_id, definition_json, state, revision,
+                                  last_event_sequence, committed)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(job_id) DO UPDATE SET
+                    state = excluded.state, revision = excluded.revision,
+                    last_event_sequence = excluded.last_event_sequence,
+                    committed = excluded.committed
+                """,
+                (job_id, definition, state, revision, last_event_sequence, int(committed)),
+            )
+            if superseded_command_ids:
+                placeholders = ",".join("?" for _ in superseded_command_ids)
+                rows = connection.execute(
+                    f"SELECT command_id, command_json FROM commands WHERE command_id IN ({placeholders})",
+                    superseded_command_ids,
+                ).fetchall()
+                owned_ids = [
+                    row["command_id"] for row in rows
+                    if _command_job_id(row["command_json"]) == job_id
+                ]
+                if owned_ids:
+                    owned_placeholders = ",".join("?" for _ in owned_ids)
+                    connection.execute(
+                        f"UPDATE commands SET state = 'superseded' WHERE command_id IN ({owned_placeholders})",
+                        owned_ids,
+                    )
+
     def begin_file_operation(
         self,
         operation_id: str,
@@ -591,6 +686,18 @@ class ClientStore:
                 connection.close()
 
 
+def _command_job_id(command_json: str) -> str:
+    """Return the job target from canonical protobuf JSON, if it has one."""
+    command = json.loads(command_json)
+    if "assignJob" in command:
+        return str(command["assignJob"].get("job", {}).get("jobId", ""))
+    if "executeStep" in command:
+        return str(command["executeStep"].get("jobId", ""))
+    if "cancelJob" in command:
+        return str(command["cancelJob"].get("jobId", ""))
+    return ""
+
+
 def _canonical(value: object) -> str:
     return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)