Add Phase 2 durable workstation agent job routing

app/modules/documents/agent_jobs.py

@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import or_, select
+from sqlalchemy.exc import IntegrityError
+
+from app.modules.documents.models import ERPAgentJob, ERPAgentJobEvent, ERPWorkstationAgent
+
+
+LEASE_SECONDS = 45
+DEFAULT_MAX_ATTEMPTS = 3
+TERMINAL_STATUSES = {"succeeded", "failed", "cancelled"}
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _json_dump(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str)
+
+
+def _json_load(value: str | None, default: Any) -> Any:
+    if not value:
+        return default
+    try:
+        parsed = json.loads(value)
+        return parsed
+    except Exception:
+        return default
+
+
+def _event(db, job: ERPAgentJob, event_type: str, *, detail: str | None = None, actor_user_id: int | None = None) -> None:
+    db.add(
+        ERPAgentJobEvent(
+            job_id=job.id,
+            tenant_id=job.tenant_id,
+            branch_id=job.branch_id,
+            workstation_agent_id=job.workstation_agent_id,
+            event_type=str(event_type or "event")[:40],
+            detail=(detail or None),
+            actor_user_id=actor_user_id,
+            occurred_at_utc=_utcnow(),
+        )
+    )
+
+
+def enqueue_agent_job(
+    db,
+    *,
+    workstation_agent_id: int,
+    action: str,
+    payload: dict[str, Any] | None = None,
+    idempotency_key: str | None = None,
+    priority: int = 5,
+    max_attempts: int = DEFAULT_MAX_ATTEMPTS,
+    created_by_user_id: int | None = None,
+) -> ERPAgentJob:
+    workstation = db.get(ERPWorkstationAgent, int(workstation_agent_id))
+    if not workstation or not workstation.is_active:
+        raise ValueError("The selected ERP workstation agent is unavailable or inactive.")
+    action_value = str(action or "").strip()
+    if not action_value:
+        raise ValueError("Agent job action is required.")
+    idem = str(idempotency_key or "").strip() or None
+    if idem:
+        existing = db.execute(
+            select(ERPAgentJob).where(
+                ERPAgentJob.workstation_agent_id == workstation.id,
+                ERPAgentJob.idempotency_key == idem,
+            )
+        ).scalar_one_or_none()
+        if existing:
+            return existing
+    job = ERPAgentJob(
+        job_uuid=uuid4().hex,
+        tenant_id=workstation.tenant_id,
+        branch_id=workstation.branch_id,
+        storage_node_id=workstation.storage_node_id,
+        workstation_agent_id=workstation.id,
+        action=action_value[:120],
+        payload_json=_json_dump(payload or {}),
+        idempotency_key=idem[:200] if idem else None,
+        status="queued",
+        priority=max(0, min(int(priority), 100)),
+        max_attempts=max(1, min(int(max_attempts), 20)),
+        attempts=0,
+        created_by_user_id=created_by_user_id,
+        created_at_utc=_utcnow(),
+        updated_at_utc=_utcnow(),
+    )
+    db.add(job)
+    try:
+        db.flush()
+    except IntegrityError:
+        db.rollback()
+        if idem:
+            existing = db.execute(
+                select(ERPAgentJob).where(
+                    ERPAgentJob.workstation_agent_id == workstation.id,
+                    ERPAgentJob.idempotency_key == idem,
+                )
+            ).scalar_one_or_none()
+            if existing:
+                return existing
+        raise
+    _event(db, job, "queued", actor_user_id=created_by_user_id)
+    return job
+
+
+def claim_jobs_for_workstation(db, *, storage_node_id: int, agent_instance_id: str, limit: int = 10) -> list[dict[str, Any]]:
+    instance_id = str(agent_instance_id or "").strip()
+    if not instance_id:
+        return []
+    workstation = db.execute(
+        select(ERPWorkstationAgent).where(
+            ERPWorkstationAgent.storage_node_id == int(storage_node_id),
+            ERPWorkstationAgent.agent_instance_id == instance_id,
+            ERPWorkstationAgent.is_active.is_(True),
+        )
+    ).scalar_one_or_none()
+    if not workstation:
+        return []
+
+    now = _utcnow()
+    # Expired claims become retryable. Jobs that have exhausted their attempts fail.
+    expired = list(
+        db.execute(
+            select(ERPAgentJob).where(
+                ERPAgentJob.workstation_agent_id == workstation.id,
+                ERPAgentJob.status == "claimed",
+                ERPAgentJob.lease_expires_at_utc.is_not(None),
+                ERPAgentJob.lease_expires_at_utc < now,
+            )
+        ).scalars().all()
+    )
+    for job in expired:
+        if int(job.attempts or 0) >= int(job.max_attempts or DEFAULT_MAX_ATTEMPTS):
+            job.status = "failed"
+            job.failed_at_utc = now
+            job.last_error = job.last_error or "Agent job lease expired after maximum retry attempts."
+            _event(db, job, "failed", detail=job.last_error)
+        else:
+            job.status = "queued"
+            job.claimed_at_utc = None
+            job.lease_expires_at_utc = None
+            _event(db, job, "lease_expired", detail="Job returned to queue for retry.")
+
+    stmt = (
+        select(ERPAgentJob)
+        .where(
+            ERPAgentJob.workstation_agent_id == workstation.id,
+            ERPAgentJob.status == "queued",
+            ERPAgentJob.attempts < ERPAgentJob.max_attempts,
+        )
+        .order_by(ERPAgentJob.priority.desc(), ERPAgentJob.created_at_utc.asc(), ERPAgentJob.id.asc())
+        .limit(max(1, min(int(limit), 25)))
+    )
+    rows = list(db.execute(stmt).scalars().all())
+    payloads: list[dict[str, Any]] = []
+    for job in rows:
+        job.status = "claimed"
+        job.attempts = int(job.attempts or 0) + 1
+        job.claimed_at_utc = now
+        job.lease_expires_at_utc = now + timedelta(seconds=LEASE_SECONDS)
+        job.updated_at_utc = now
+        _event(db, job, "claimed", detail=f"attempt={job.attempts}")
+        payloads.append(
+            {
+                "job_uuid": job.job_uuid,
+                "action": job.action,
+                "payload": _json_load(job.payload_json, {}),
+                "idempotency_key": job.idempotency_key or job.job_uuid,
+                "attempt": int(job.attempts or 0),
+                "max_attempts": int(job.max_attempts or DEFAULT_MAX_ATTEMPTS),
+                "lease_seconds": LEASE_SECONDS,
+            }
+        )
+    return payloads
+
+
+def complete_agent_job(
+    db,
+    *,
+    storage_node_id: int,
+    agent_instance_id: str,
+    message: dict[str, Any],
+) -> ERPAgentJob | None:
+    instance_id = str(agent_instance_id or "").strip()
+    job_uuid = str(message.get("job_uuid") or "").strip()
+    if not instance_id or not job_uuid:
+        return None
+    workstation = db.execute(
+        select(ERPWorkstationAgent).where(
+            ERPWorkstationAgent.storage_node_id == int(storage_node_id),
+            ERPWorkstationAgent.agent_instance_id == instance_id,
+            ERPWorkstationAgent.is_active.is_(True),
+        )
+    ).scalar_one_or_none()
+    if not workstation:
+        return None
+    job = db.execute(
+        select(ERPAgentJob).where(
+            ERPAgentJob.job_uuid == job_uuid,
+            ERPAgentJob.workstation_agent_id == workstation.id,
+        )
+    ).scalar_one_or_none()
+    if not job:
+        return None
+    if job.status in TERMINAL_STATUSES:
+        return job
+
+    now = _utcnow()
+    ok = bool(message.get("ok"))
+    job.result_json = _json_dump(message.get("result") or {}) if ok else None
+    job.last_error = None if ok else str(message.get("error") or "Agent job failed.")[:4000]
+    job.agent_completed_at_utc = now
+    job.updated_at_utc = now
+    job.lease_expires_at_utc = None
+    if ok:
+        job.status = "succeeded"
+        job.completed_at_utc = now
+        _event(db, job, "succeeded")
+    elif int(job.attempts or 0) < int(job.max_attempts or DEFAULT_MAX_ATTEMPTS) and bool(message.get("retryable", False)):
+        job.status = "queued"
+        job.claimed_at_utc = None
+        _event(db, job, "retry_queued", detail=job.last_error)
+    else:
+        job.status = "failed"
+        job.failed_at_utc = now
+        _event(db, job, "failed", detail=job.last_error)
+    return job
+
+
+def cancel_agent_job(db, job: ERPAgentJob, *, actor_user_id: int | None = None, reason: str | None = None) -> ERPAgentJob:
+    if job.status in TERMINAL_STATUSES:
+        return job
+    job.status = "cancelled"
+    job.cancelled_at_utc = _utcnow()
+    job.updated_at_utc = job.cancelled_at_utc
+    job.lease_expires_at_utc = None
+    _event(db, job, "cancelled", detail=reason, actor_user_id=actor_user_id)
+    return job
+
+
+def list_recent_agent_jobs(db, *, tenant_id: int | None = None, branch_id: int | None = None, limit: int = 25) -> list[ERPAgentJob]:
+    stmt = select(ERPAgentJob).order_by(ERPAgentJob.created_at_utc.desc(), ERPAgentJob.id.desc())
+    if tenant_id is not None:
+        stmt = stmt.where(ERPAgentJob.tenant_id == int(tenant_id))
+    if branch_id is not None:
+        stmt = stmt.where(ERPAgentJob.branch_id == int(branch_id))
+    stmt = stmt.limit(max(1, min(int(limit), 200)))
+    return list(db.execute(stmt).scalars().all())

app/modules/documents/agent_package.py

@@ -4,7 +4,7 @@ import io
 from pathlib import Path
 import zipfile
 
-ERP_LOCAL_AGENT_VERSION = "1.10.0"
+ERP_LOCAL_AGENT_VERSION = "1.11.0"
 ERP_LOCAL_AGENT_NAME = "ERP Local Agent"
 RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime"
 _DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0)

app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "1.10.0"
+__version__ = "1.11.0"
 AGENT_NAME = "ERP Local Agent"

app/modules/documents/local_agent_runtime/erp_local_agent/db.py

@@ -71,6 +71,16 @@ class LocalDB:
                     status TEXT NOT NULL DEFAULT 'received',
                     error_message TEXT NULL
                 );
+                CREATE TABLE IF NOT EXISTS processed_agent_jobs (
+                    job_uuid TEXT PRIMARY KEY,
+                    idempotency_key TEXT NOT NULL UNIQUE,
+                    action TEXT NOT NULL,
+                    execution_status TEXT NOT NULL DEFAULT 'processing',
+                    ok INTEGER NULL,
+                    result_json TEXT NULL,
+                    error_message TEXT NULL,
+                    processed_at_utc TEXT NOT NULL
+                );
                 CREATE TABLE IF NOT EXISTS update_history (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
                     event_type TEXT NOT NULL,
@@ -192,6 +202,42 @@ class LocalDB:
             )
             conn.commit()
 
+    def get_processed_agent_job(self, job_uuid: str, idempotency_key: str) -> dict[str, Any] | None:
+        with self.connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM processed_agent_jobs WHERE job_uuid=? OR idempotency_key=? ORDER BY processed_at_utc DESC LIMIT 1",
+                (str(job_uuid or ""), str(idempotency_key or "")),
+            ).fetchone()
+        if not row:
+            return None
+        result = dict(row)
+        try:
+            result["result"] = json.loads(result.get("result_json") or "{}")
+        except Exception:
+            result["result"] = {}
+        return result
+
+    def begin_agent_job(self, job_uuid: str, idempotency_key: str, action: str) -> None:
+        with self.connect() as conn:
+            conn.execute(
+                """INSERT OR IGNORE INTO processed_agent_jobs
+                   (job_uuid, idempotency_key, action, execution_status, ok, result_json, error_message, processed_at_utc)
+                   VALUES (?, ?, ?, 'processing', NULL, NULL, NULL, ?)""",
+                (str(job_uuid), str(idempotency_key), str(action), _utc_now()),
+            )
+            conn.commit()
+
+    def record_processed_agent_job(self, job_uuid: str, idempotency_key: str, action: str, ok: bool, result: Any = None, error_message: str | None = None) -> None:
+        result_json = None if result is None else json.dumps(result, ensure_ascii=False, default=str)
+        with self.connect() as conn:
+            conn.execute(
+                """INSERT OR REPLACE INTO processed_agent_jobs
+                   (job_uuid, idempotency_key, action, execution_status, ok, result_json, error_message, processed_at_utc)
+                   VALUES (?, ?, ?, 'completed', ?, ?, ?, ?)""",
+                (str(job_uuid), str(idempotency_key), str(action), 1 if ok else 0, result_json, error_message, _utc_now()),
+            )
+            conn.commit()
+
     def record_update(self, event_type: str, from_version: str, to_version: str, status: str = "", sha256: str = "", package_path: str = "", error_message: str | None = None, details: Any = None) -> None:
         details_json = None if details is None else json.dumps(details, ensure_ascii=False, default=str)
         with self.connect() as conn:

app/modules/documents/local_agent_runtime/erp_local_agent/tunnel.py

@@ -71,6 +71,7 @@ class StorageAgentTunnel:
                     jobs = message.get("jobs") or []
                     requests_ = message.get("download_requests") or message.get("requests") or []
                     commands = message.get("commands") or []
+                    agent_jobs = message.get("agent_jobs") or []
                     await asyncio.to_thread(self.agent.process_storage_jobs_from_payload, jobs)
                     await asyncio.to_thread(self.agent.process_download_requests_from_payload, requests_)
                     for command in commands:
@@ -80,11 +81,14 @@ class StorageAgentTunnel:
                         result = await asyncio.to_thread(self.command_processor.process, command)
                         self.db.command_completed(history_id, bool(result.get("ok")), result.get("error"))
                         await websocket.send(self._json_text(result))
+                    for job in agent_jobs:
+                        await self._process_agent_job(websocket, job)
                     await websocket.send(self._json_text({
                         "type": "agent_status",
                         "jobs_seen": len(jobs),
                         "requests_seen": len(requests_),
                         "commands_seen": len(commands),
+                        "agent_jobs_seen": len(agent_jobs),
                         "capabilities": self._capabilities(),
                         "workstation": self._workstation_status(),
                         "agent_time_utc": self._now(),
@@ -93,9 +97,100 @@ class StorageAgentTunnel:
                 if msg_type == "error":
                     self.logger.error("Tunnel server error: %s", message.get("error"))
 
+    async def _process_agent_job(self, websocket, job: dict[str, Any]) -> None:
+        job_uuid = str(job.get("job_uuid") or "").strip()
+        idempotency_key = str(job.get("idempotency_key") or job_uuid).strip()
+        action = str(job.get("action") or "").strip()
+        if not job_uuid or not idempotency_key or not action:
+            return
+
+        cached = self.db.get_processed_agent_job(job_uuid, idempotency_key)
+        if cached:
+            if str(cached.get("execution_status") or "").lower() == "processing":
+                await websocket.send(self._json_text({
+                    "type": "agent_job_result",
+                    "job_uuid": job_uuid,
+                    "idempotency_key": idempotency_key,
+                    "ok": False,
+                    "result": {},
+                    "error": "Previous execution was interrupted before acknowledgement. Verify Tally before retrying this job.",
+                    "retryable": False,
+                    "indeterminate": True,
+                    "replayed": True,
+                    "agent_time_utc": self._now(),
+                    "workstation": self._workstation_status(),
+                }))
+                return
+            await websocket.send(self._json_text({
+                "type": "agent_job_result",
+                "job_uuid": job_uuid,
+                "idempotency_key": idempotency_key,
+                "ok": bool(cached.get("ok")),
+                "result": cached.get("result") or {},
+                "error": cached.get("error_message"),
+                "retryable": False,
+                "replayed": True,
+                "agent_time_utc": self._now(),
+                "workstation": self._workstation_status(),
+            }))
+            return
+
+        # Persist 'processing' before touching Tally. If Windows/Tally/agent dies in
+        # the narrow post-before-ack window, the next connection refuses to blindly
+        # execute the same accounting job again.
+        self.db.begin_agent_job(job_uuid, idempotency_key, action)
+        history_id = self.db.command_received(job_uuid, action)
+        command = {
+            "command_id": job_uuid,
+            "action": action,
+            "payload": job.get("payload") or {},
+        }
+        try:
+            result = await asyncio.to_thread(self.command_processor.process, command)
+            ok = bool(result.get("ok"))
+            error = result.get("error")
+            result_payload = result.get("result") or {}
+            self.db.command_completed(history_id, ok, error)
+            # Both success and deterministic application errors are cached so a
+            # reconnect cannot execute the same accounting write twice.
+            self.db.record_processed_agent_job(
+                job_uuid, idempotency_key, action, ok, result_payload, error
+            )
+            await websocket.send(self._json_text({
+                "type": "agent_job_result",
+                "job_uuid": job_uuid,
+                "idempotency_key": idempotency_key,
+                "ok": ok,
+                "result": result_payload,
+                "error": error,
+                "retryable": False,
+                "replayed": False,
+                "agent_time_utc": self._now(),
+                "workstation": self._workstation_status(),
+            }))
+        except Exception as exc:
+            error = str(exc)
+            self.db.command_completed(history_id, False, error)
+            # Execution has already crossed the durable local 'processing' marker.
+            # We cannot prove that Tally did not accept a write before this exception,
+            # therefore this is deliberately non-retryable and requires verification.
+            await websocket.send(self._json_text({
+                "type": "agent_job_result",
+                "job_uuid": job_uuid,
+                "idempotency_key": idempotency_key,
+                "ok": False,
+                "result": {},
+                "error": f"Agent execution became indeterminate: {error}",
+                "retryable": False,
+                "indeterminate": True,
+                "replayed": False,
+                "agent_time_utc": self._now(),
+                "workstation": self._workstation_status(),
+            }))
+
     def _capabilities(self) -> list[str]:
         return [
-            "storage", "local_dashboard", "manual_updates", "accounting_act",
+            "storage", "local_dashboard", "manual_updates", "accounting_act", "agent.jobs.v2",
             "tally.status", "tally.company_identity", "tally.read_groups",
             "tally.read_ledgers", "tally.read_stock_items", "tally.read_vouchers",
             "tally.read_trial_balance", "tally.master_sync", "tally.transaction_sync",

app/modules/documents/models.py

@@ -218,6 +218,62 @@ class ERPWorkstationAgent(CommonBase):
     storage_node = relationship("BranchStorageNode")
 
 
+class ERPAgentJob(CommonBase):
+    """Durable server-side job routed to one specific ERP workstation agent."""
+
+    __tablename__ = "erp_agent_jobs"
+    __table_args__ = (
+        UniqueConstraint("job_uuid", name="uq_erp_agent_jobs_uuid"),
+        UniqueConstraint("workstation_agent_id", "idempotency_key", name="uq_erp_agent_jobs_workstation_idempotency"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    job_uuid: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    branch_id: Mapped[int | None] = mapped_column(ForeignKey("branches.id", ondelete="SET NULL"), nullable=True, index=True)
+    storage_node_id: Mapped[int] = mapped_column(ForeignKey("branch_storage_nodes.id", ondelete="CASCADE"), nullable=False, index=True)
+    workstation_agent_id: Mapped[int] = mapped_column(ForeignKey("erp_workstation_agents.id", ondelete="CASCADE"), nullable=False, index=True)
+    action: Mapped[str] = mapped_column(String(120), nullable=False, index=True)
+    payload_json: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
+    idempotency_key: Mapped[str | None] = mapped_column(String(200), nullable=True, index=True)
+    status: Mapped[str] = mapped_column(String(30), nullable=False, default="queued", index=True)
+    priority: Mapped[int] = mapped_column(Integer, nullable=False, default=5, index=True)
+    max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
+    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    result_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    created_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False, index=True)
+    updated_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False)
+    claimed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    lease_expires_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    agent_completed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    completed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    failed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    cancelled_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+
+    workstation = relationship("ERPWorkstationAgent")
+    storage_node = relationship("BranchStorageNode")
+
+
+class ERPAgentJobEvent(CommonBase):
+    """Immutable audit event for an ERP workstation-agent job."""
+
+    __tablename__ = "erp_agent_job_events"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    job_id: Mapped[int] = mapped_column(ForeignKey("erp_agent_jobs.id", ondelete="CASCADE"), nullable=False, index=True)
+    tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    branch_id: Mapped[int | None] = mapped_column(ForeignKey("branches.id", ondelete="SET NULL"), nullable=True, index=True)
+    workstation_agent_id: Mapped[int] = mapped_column(ForeignKey("erp_workstation_agents.id", ondelete="CASCADE"), nullable=False, index=True)
+    event_type: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
+    detail: Mapped[str | None] = mapped_column(Text, nullable=True)
+    actor_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    occurred_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False, index=True)
+
+    job = relationship("ERPAgentJob")
+
+
 class DocumentStorageJob(CommonBase):
     """Cloud-to-local branch storage transfer job.
 

app/modules/documents/templates/documents/storage_nodes.html

@@ -177,7 +177,7 @@
         </div>
         <div class="overflow-x-auto">
           <table class="w-full text-sm">
-            <thead class="bg-slate-50 text-slate-600"><tr><th class="p-3 text-left">Workstation</th><th class="p-3 text-left">Agent</th><th class="p-3 text-left">Tally</th><th class="p-3 text-left">Open Companies</th><th class="p-3 text-left">Last Seen</th></tr></thead>
+            <thead class="bg-slate-50 text-slate-600"><tr><th class="p-3 text-left">Workstation</th><th class="p-3 text-left">Agent</th><th class="p-3 text-left">Tally</th><th class="p-3 text-left">Open Companies</th><th class="p-3 text-left">Last Seen</th><th class="p-3 text-left">Phase 2</th></tr></thead>
             <tbody>
             {% for ws in workstation_agents or [] %}
               <tr class="border-t align-top">
@@ -186,9 +186,44 @@
                 <td class="p-3"><span class="px-2 py-1 rounded-full text-xs {{ 'bg-emerald-100 text-emerald-700' if ws.tally_connected else 'bg-slate-100 text-slate-600' }}">{{ 'Connected' if ws.tally_connected else 'Not connected' }}</span></td>
                 <td class="p-3">{{ ws.tally_company_count or 0 }}</td>
                 <td class="p-3 text-xs">{{ ws.last_seen_at_utc or '-' }}</td>
+                <td class="p-3">
+                  {% if can_manage_branch_storage %}
+                  <form method="post" action="/documents/storage-workstations/{{ ws.id }}/test-route">
+                    <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
+                    <button class="text-xs font-semibold text-blue-700">Test route</button>
+                  </form>
+                  {% else %}-{% endif %}
+                </td>
               </tr>
             {% else %}
-              <tr><td colspan="5" class="p-6 text-center text-slate-500">No workstation agent has connected since the Phase 1 upgrade.</td></tr>
+              <tr><td colspan="6" class="p-6 text-center text-slate-500">No workstation agent has connected since the Phase 1 upgrade.</td></tr>
+            {% endfor %}
+            </tbody>
+          </table>
+        </div>
+      </div>
+
+
+      <div class="rounded-2xl border bg-white overflow-hidden shadow-sm">
+        <div class="p-4 border-b">
+          <h2 class="font-bold text-slate-900">Recent Workstation Agent Jobs</h2>
+          <p class="text-xs text-slate-500 mt-1">Phase 2 jobs are durable, routed to one workstation, leased for safe retry, and protected by an idempotency key.</p>
+        </div>
+        <div class="overflow-x-auto">
+          <table class="w-full text-xs">
+            <thead class="bg-slate-50 text-slate-600"><tr><th class="p-2 text-left">Job</th><th class="p-2 text-left">Workstation</th><th class="p-2 text-left">Action</th><th class="p-2 text-left">Status</th><th class="p-2 text-left">Attempts</th><th class="p-2 text-left">Created</th></tr></thead>
+            <tbody>
+            {% for job in recent_agent_jobs or [] %}
+              <tr class="border-t">
+                <td class="p-2 font-mono">{{ job.job_uuid[:10] }}...</td>
+                <td class="p-2">{{ job.workstation.machine_name if job.workstation else ('#' ~ job.workstation_agent_id) }}</td>
+                <td class="p-2">{{ job.action }}</td>
+                <td class="p-2"><span class="font-semibold">{{ job.status }}</span>{% if job.last_error %}<div class="text-red-600 mt-1">{{ job.last_error }}</div>{% endif %}</td>
+                <td class="p-2">{{ job.attempts }}/{{ job.max_attempts }}</td>
+                <td class="p-2">{{ job.created_at_utc or '-' }}</td>
+              </tr>
+            {% else %}
+              <tr><td colspan="6" class="p-4 text-center text-slate-500">No Phase 2 workstation jobs yet. Use Test route on a connected workstation to verify routing.</td></tr>
             {% endfor %}
             </tbody>
           </table>

app/modules/documents/ui.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import date, datetime, timezone
+from uuid import uuid4
 import json
 import asyncio
 import logging
@@ -83,6 +84,7 @@ from app.modules.core.tenancy.models import Branch, Tenant
 from app.modules.core.tenancy.year_control import is_row_financial_year_locked
 from app.modules.documents.agent_package import ERP_LOCAL_AGENT_VERSION, build_agent_env, build_preconfigured_agent_zip, build_agent_update_zip
 from app.modules.documents.models import BranchStorageNode, ERPWorkstationAgent
+from app.modules.documents.agent_jobs import claim_jobs_for_workstation, complete_agent_job, list_recent_agent_jobs, enqueue_agent_job
 from app.modules.accounting.agent_bridge import list_pending_agent_commands, record_agent_command_result
 
 from app.modules.services.task_documents import (
@@ -1319,6 +1321,7 @@ def storage_nodes(request: Request):
         jobs = list_storage_jobs(db, tenant_id=tenant_filter, branch_id=branch_filter, limit=10)
         requests = list_download_requests(db, tenant_id=tenant_filter, branch_id=branch_filter, limit=10)
         workstations = _visible_workstation_agents(db, tenant_filter, branch_filter)
+        agent_jobs = list_recent_agent_jobs(db, tenant_id=tenant_filter, branch_id=branch_filter, limit=20)
         return _render(
             request,
             "modules/documents/templates/documents/storage_nodes.html",
@@ -1331,6 +1334,7 @@ def storage_nodes(request: Request):
             recent_jobs=jobs,
             recent_download_requests=requests,
             workstation_agents=workstations,
+            recent_agent_jobs=agent_jobs,
             generated_secret=None,
             storage_scope_title=_storage_scope_title(user, scope),
             erp_local_agent_version=ERP_LOCAL_AGENT_VERSION,
@@ -1341,6 +1345,43 @@ def storage_nodes(request: Request):
         db.close()
 
 
+@router.post("/storage-workstations/{workstation_id}/test-route")
+def test_workstation_job_route(request: Request, workstation_id: int, csrf_token: str = Form(...)):
+    """Queue a harmless workstation-specific Tally status job for Phase 2 routing verification."""
+    validate_csrf(request, csrf_token)
+    db = CommonSessionLocal()
+    try:
+        user, response = _require_user(request, db, "documents.upload")
+        if response:
+            return response
+        scope = build_document_scope(request, db, user)
+        if not _can_manage_branch_storage(scope):
+            return _redirect_denied()
+        workstation = db.get(ERPWorkstationAgent, int(workstation_id))
+        if not workstation:
+            return _redirect_denied()
+        tenant_filter = _storage_tenant_filter(user, scope)
+        branch_filter = _storage_branch_filter(user, scope)
+        if tenant_filter is not None and int(workstation.tenant_id) != int(tenant_filter):
+            return _redirect_denied()
+        if branch_filter is not None and int(workstation.branch_id or 0) != int(branch_filter):
+            return _redirect_denied()
+        enqueue_agent_job(
+            db,
+            workstation_agent_id=workstation.id,
+            action="tally_status",
+            payload={},
+            idempotency_key=f"route-test:{workstation.id}:{uuid4().hex}",
+            created_by_user_id=int(user.id),
+            priority=10,
+            max_attempts=3,
+        )
+        db.commit()
+        return RedirectResponse(url="/documents/storage-nodes", status_code=303)
+    finally:
+        db.close()
+
+
 @router.get("/branch-storage-dashboard")
 def partner_branch_storage_dashboard(request: Request):
     """Partner/Branch dashboard card for local storage setup and status.
@@ -1737,15 +1778,28 @@ def _permanent_download_requests_payload(items):
     ]
 
 
-def _storage_agent_sync_payload(db, node):
+def _storage_agent_sync_payload(db, node, workstation_instance_id: str | None = None):
     # Retry cleanup in tunnel mode as well as normal heartbeat mode.
     retry_verified_vps_cleanup_for_node(db, node, limit=25)
     jobs = _normal_storage_jobs_payload(list_pending_storage_jobs(db, node))
     jobs += _permanent_storage_jobs_payload(list_pending_permanent_storage_jobs(db, node))
     requests_ = _normal_download_requests_payload(list_pending_download_requests(db, node))
     requests_ += _permanent_download_requests_payload(list_pending_permanent_download_requests(db, node))
+    # Legacy node-level commands are retained unchanged for backward compatibility.
     commands = list_pending_agent_commands(node.node_code, limit=10)
-    return {"ok": True, "jobs": jobs, "download_requests": requests_, "requests": requests_, "commands": commands}
+    agent_jobs = []
+    if workstation_instance_id:
+        agent_jobs = claim_jobs_for_workstation(
+            db, storage_node_id=node.id, agent_instance_id=workstation_instance_id, limit=10
+        )
+    return {
+        "ok": True,
+        "jobs": jobs,
+        "download_requests": requests_,
+        "requests": requests_,
+        "commands": commands,
+        "agent_jobs": agent_jobs,
+    }
 
 
 @router.get("/storage-agent/jobs/pending")
@@ -1875,6 +1929,7 @@ async def storage_agent_tunnel(websocket: WebSocket):
         db.close()
 
     last_push = 0.0
+    connection_workstation_instance_id = ""
     while True:
         try:
             try:
@@ -1891,10 +1946,21 @@ async def storage_agent_tunnel(websocket: WebSocket):
                     return
                 if message and isinstance(message.get("workstation"), dict):
                     client_ip = websocket.client.host if websocket.client else None
-                    _record_workstation_status(db, node, message.get("workstation"), client_ip)
+                    workstation_payload = message.get("workstation")
+                    _record_workstation_status(db, node, workstation_payload, client_ip)
+                    advertised_instance_id = str(workstation_payload.get("agent_instance_id") or "").strip()
+                    if advertised_instance_id:
+                        connection_workstation_instance_id = advertised_instance_id
                 if message and message.get("type") == "command_result":
                     record_agent_command_result(node.node_code, message)
-                if message and message.get("type") in {"heartbeat", "agent_status", "command_result"}:
+                if message and message.get("type") == "agent_job_result":
+                    complete_agent_job(
+                        db,
+                        storage_node_id=node.id,
+                        agent_instance_id=connection_workstation_instance_id,
+                        message=message,
+                    )
+                if message and message.get("type") in {"heartbeat", "agent_status", "command_result", "agent_job_result"}:
                     try:
                         node.storage_mode = "tunnel"
                     except Exception:
@@ -1902,7 +1968,7 @@ async def storage_agent_tunnel(websocket: WebSocket):
                 now = datetime.now(timezone.utc).timestamp()
                 force = bool(message and message.get("type") in {"ready", "sync_now", "agent_status"})
                 if force or now - last_push >= 5:
-                    payload = _storage_agent_sync_payload(db, node)
+                    payload = _storage_agent_sync_payload(db, node, connection_workstation_instance_id)
                     payload.update({"type": "sync", "node_code": node.node_code})
                     await websocket.send_json(payload)
                     last_push = now