refactor(agents): route workflow agents through Hub sessions

Author: alexeygrigorev

README.md

@@ -388,15 +388,10 @@ allow-list ignores everyone. `mode=anyone` still refuses a message with no
 From. Additional workflow filters AND with the address rule. Sender admission
 is separate from receiving-address configuration and outbound senders.
 
-An `agent` action queues a prompt for a Dapier worker — a machine running
-`dapier worker`, which picks the task up and runs a headless Claude Code
-process there, reporting the terminal result through the authenticated API.
-Tasks stay queued until a worker is running; the console's **Workers** tab
-and `dapier workers list` show which workers are active. An agent action can
-sit on an email, a webhook, a schedule, or a poll.
-The host needs a dedicated Dapier token, not AWS credentials or Aplexer; see
-[headless worker setup](docs/headless-worker.md). Webhooks are managed with
-`dapier webhooks`. A sample of what a flow receives is `dapier workflows sample`.
+The `hub_session` action creates a Hub intake and requests a Hub-managed
+headless session. Use it for workflow-triggered coding work; Dapier does not
+run coding agents on host workers. Webhooks are managed with `dapier webhooks`.
+A sample of what a flow receives is `dapier workflows sample`.
 
 A receiving address is a workflow trigger, for example:
 
@@ -417,7 +412,7 @@ trigger:
 
 Action types and their keys match the workflow catalog (`webhook`, `slack`,
 `telegram_send`, `email_send`, `dataops`, `dropbox_upload`, `dropbox_delete`,
-`render_html_to_pdf`, `code`). Text fields accept `{field}` templates from the
+`render_html_to_pdf`, `code`, `hub_session`). Text fields accept `{field}` templates from the
 triggering event; `{trigger.field}` reaches the same data (plus envelope
 scalars like `trigger.connector` and `trigger.occurred_at`), and
 `{steps.<action_id>.output.<path>}` / `{steps.<action_id>.status}` reference an

dapier_cli/cli/ops.py

@@ -1,10 +1,10 @@
-"""Single-command operator nouns: overview, usage, quota, errors, agent-tasks, workers, worker, catalog."""
+"""Single-command operator nouns: overview, usage, quota, errors, catalog."""
 
 import sys
 
 from .. import commands
 
-GROUPS = ("costs", "overview", "usage", "quota", "errors", "agent-tasks", "workers", "worker", "catalog",
+GROUPS = ("costs", "overview", "usage", "quota", "errors", "catalog",
           "store-check",)
 
 
@@ -29,40 +29,6 @@ def register(sub):
     errors_sub = errors_p.add_subparsers(dest="command")
     errors_sub.add_parser("send-digest",
                           help="Render and email the error digest now (operator)")
-    agent_tasks_p = sub.add_parser(
-        "agent-tasks", help="Host jobs the agent action enqueued (run by `dapier worker`)")
-    agent_tasks_sub = agent_tasks_p.add_subparsers(dest="command", required=True)
-    agent_tasks_list = agent_tasks_sub.add_parser("list", help="List host agent tasks, newest first")
-    agent_tasks_list.add_argument("--limit", default=None,
-                                  help="How many to show (default 50, max 200)")
-    agent_tasks_list.add_argument("--status", default=None,
-                                  help="Only rows with this status: queued, running, succeeded, failed, timed_out, interrupted")
-    agent_tasks_list.add_argument("--owner",
-                                  help="Only this person's tasks (alias or email, for example orin)")
-    agent_tasks_list.add_argument("--as", dest="as_id",
-                                  help="Only tasks for this subject id")
-    agent_tasks_list.add_argument("--assignee",
-                                  help="Only tasks assigned to this subject id")
-    agent_tasks_show = agent_tasks_sub.add_parser("show", help="Show one agent task and its recorded logs")
-    agent_tasks_show.add_argument("task_id")
-    workers_p = sub.add_parser(
-        "workers", help="Host workers (`dapier worker` processes) that run agent tasks")
-    workers_sub = workers_p.add_subparsers(dest="command", required=True)
-    workers_sub.add_parser("list", help="List host workers, most recently seen first")
-    worker_p = sub.add_parser(
-        "worker",
-        help="Run headless host jobs over HTTPS until interrupted.")
-    worker_p.add_argument("--token-file", default=None,
-                          help="Owner-only Dapier host-worker token file")
-    worker_p.add_argument("--workspace-root", default=None,
-                          help="Host job root directory (default ~/dapier-ws)")
-    worker_p.add_argument("--max-runtime", type=int, default=3600,
-                          help="Maximum seconds per headless job")
-    worker_p.add_argument("--capability", action="append", default=[],
-                          help="Advertise a capability, e.g. browser (repeatable)")
-    worker_p.add_argument("--engine", choices=("claude", "codex"), default="claude",
-                          help="Agent engine available on this worker")
-    worker_p.add_argument("--once", action="store_true", help="Poll once and exit")
     catalog_p = sub.add_parser("catalog", help="Show the action and trigger catalog (GET /api/catalog)")
     catalog_p.add_argument("--json", action="store_true", help="Print the raw catalog JSON")
     store_check_p = sub.add_parser(
@@ -89,22 +55,6 @@ def run(args, api_url, debug, child=None):
         if getattr(args, "command", None) == "send-digest":
             return commands.errors_send_digest(api_url, debug)
         return commands.errors_summary(api_url, debug, days=args.days)
-    if args.group == "agent-tasks":
-        if args.command == "show":
-            return commands.agent_tasks_show(api_url, args.task_id, debug)
-        return commands.agent_tasks_list(api_url, debug,
-                                         limit=getattr(args, "limit", None),
-                                         status=getattr(args, "status", None),
-                                         owner=getattr(args, "owner", None),
-                                         as_id=getattr(args, "as_id", None),
-                                         assignee=getattr(args, "assignee", None))
-    if args.group == "workers":
-        return commands.workers_list(api_url, debug)
-    if args.group == "worker":
-        return commands.worker_run(api_url, token_file=args.token_file,
-                                   workspace_root=args.workspace_root,
-                                   max_runtime=args.max_runtime, once=args.once,
-                                   capabilities=args.capability, engine=args.engine)
     if args.group == "catalog":
         return commands.catalog_show(api_url, debug, as_json=args.json)
     if args.group == "store-check":

dapier_cli/commands/__init__.py

@@ -25,7 +25,6 @@
 from .storage import *
 from .oauth_clients import *
 from .emails import *
-from .tasks import *
 from .typefully import *
 from .slack import *
 from .gdocs import *

dapier_cli/commands/tasks.py

@@ -37,8 +37,7 @@ def tokens_create(api_url, name, agent, debug=False, output=None):
                     {"token_id": name, "agent": agent}, debug=debug)
     print(f"Created API token {data.get('token_id')} "
           f"(subject {data.get('subject')}, agent {data.get('agent')}).")
-    if agent != "host-worker":
-        print(f"Grant it access with `dapier grants save` using subject {data.get('subject')}.")
+    print(f"Grant it access with `dapier grants save` using subject {data.get('subject')}.")
     if output:
         path = Path(output).expanduser()
         path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)

dapier_cli/commands/tokens.py

@@ -129,20 +129,6 @@ export const logicOperators = [
 
 /** Mirrors the run_* dispatch in src/dapier/engine/__init__.py. */
 export const actionCatalog: ActionEntry[] = [
-  {
-    type: "agent",
-    label: "Agent",
-    icon: Bot,
-    description: "Queue a prompt for a Dapier worker — a machine running `dapier worker`, listed on the console's Workers page. The worker runs the prompt as a headless Claude session in its workspace; the step returns as soon as the task is queued. With no worker running, tasks stay queued until one starts. The trigger's stored attachments (an email's files) are staged into the workspace under attachments/ and the run is told where they landed; set Pass trigger attachments to off to skip them.",
-    fields: [
-      { key: "prompt", label: "Prompt", type: "textarea", required: true, placeholder: "{subject}\n\n{text}" },
-      { key: "workspace", label: "Workspace", required: true, placeholder: "/home/alexey/git/dapier" },
-      { key: "engine", label: "Engine", placeholder: "claude" },
-      { key: "requires", label: "Required capabilities", placeholder: "browser" },
-      { key: "tag_prefix", label: "Tag prefix", placeholder: "agent" },
-      { key: "attachments", label: "Pass trigger attachments", type: "select", options: ["default", "off"], default: "default" }
-    ]
-  },
   {
     type: "hub_session",
     label: "Hub session",