fix: claim/unclaim use view endpoint so tasks actually move buckets, log review skips at INFO

Reviewed-on: #2

.woodpecker.yaml

@@ -54,6 +54,27 @@ steps:
         branch: mainline
       - event: tag
 
+  # ── Intake image ─────────────────────────────────────────────────────────────
+  - name: build-intake
+    image: woodpeckerci/plugin-docker-buildx
+    settings:
+      registry: registry.ctz.fyi
+      repo: registry.ctz.fyi/library/autojanet-intake
+      dockerfile: intake/Dockerfile
+      context: intake/
+      username:
+        from_secret: RS_HARBOR_USER
+      password:
+        from_secret: RS_HARBOR_PASS
+      tags:
+        - latest
+        - "${CI_COMMIT_SHA:0:12}"
+      platforms: linux/amd64
+    when:
+      - event: push
+        branch: mainline
+      - event: tag
+
   # ── Trivy scan agent image ───────────────────────────────────────────────────
   - name: trivy-agent
     image: aquasec/trivy:latest

container/entrypoint.py

@@ -164,6 +164,12 @@ Instructions:
 5. Leave a comment on the Vikunja task summarising what you did and linking any PR.
 6. Do not mark the task as done — the entrypoint will move it to In Review when you finish.
 7. Do not ask for confirmation — act autonomously within your role constraints.
+
+IMPORTANT — avoid looping:
+- If you attempt the same fix more than twice and it is still not working, STOP.
+- Do not keep retrying the same approach hoping for a different result.
+- Instead: leave a Vikunja comment explaining exactly what you tried and what is blocking you, then exit.
+- A human will read it and unblock you. Spinning wastes time and money.
 """
 
 
@@ -193,6 +199,25 @@ def move_task_to_bucket(pm_token: str, bucket_id: str) -> None:
         log.warning("Failed to move task to bucket %s: %s", bucket_id, e)
 
 
+def post_task_comment(pm_token: str, rc: int) -> None:
+    """Post a completion comment on the Vikunja task."""
+    if rc == 0:
+        body = f"✅ **{AGENT_ROLE}** agent completed this task and moved it to In Review."
+    else:
+        body = f"⚠️ **{AGENT_ROLE}** agent finished with exit code {rc}. Task moved to In Review for manual check."
+    try:
+        resp = httpx.put(
+            f"{VIKUNJA_BASE_URL}/api/v1/tasks/{TASK_ID}/comments",
+            headers={"Authorization": f"Bearer {pm_token}", "Content-Type": "application/json"},
+            json={"comment": body},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        log.info("Posted completion comment on task %s", TASK_ID)
+    except Exception as e:
+        log.warning("Failed to post task comment: %s", e)
+
+
 def main() -> None:
     log.info("Agent entrypoint: role=%s task=%s", AGENT_ROLE, TASK_ID)
 
@@ -203,7 +228,7 @@ def main() -> None:
     write_opencode_config(secrets, AGENT_ROLE)
     configure_git_identity()
 
-    # Fetch pm token for bucket moves (admin operation)
+    # Fetch pm token for bucket moves and comments (admin operation)
     pm_token = ""
     try:
         pm_token = get_secret(bao_token, "autojanet/pm/vikunja-token", "token")
@@ -214,8 +239,9 @@ def main() -> None:
     prompt = build_prompt(TASK_ID, TASK_TITLE)
     rc = run_opencode(prompt)
 
-    # Move to In Review regardless of exit code — work was attempted
+    # Always post comment and move bucket regardless of exit code
     if pm_token:
+        post_task_comment(pm_token, rc)
         move_task_to_bucket(pm_token, IN_REVIEW_BUCKET_ID)
 
     if rc != 0:

dispatcher/dispatcher.py

@@ -3,9 +3,13 @@
 AutoJanet Dispatcher
 
 Runs as a CronJob every 2 minutes. Discovers the target Vikunja project by
-name, resolves the kanban view and all 5 standard bucket IDs by name, then
-claims tasks from the Todo bucket that have an `agent:<role>` label and
-spawns a Kubernetes Job for the appropriate agent.
+name, resolves the kanban view and all 5 standard bucket IDs by name, then:
+
+  1. Sweeps done=True tasks into the Done bucket.
+  2. Watchdog: moves stale InProgress tasks (dead job) back to Todo.
+  3. Review orchestration: spawns a PM agent for each task in Review that
+     needs sub-tasks created (code-review, security-review, etc.).
+  4. Claims tasks from the Todo bucket and spawns agent Jobs.
 
 Config (env vars):
   OPENBAO_ADDR, OPENBAO_ROLE_ID, OPENBAO_SECRET_ID
@@ -42,11 +46,16 @@ VIKUNJA_PROJECT_NAME  = os.environ.get("VIKUNJA_PROJECT_NAME", "Autonomous Agent
 K8S_NAMESPACE = os.environ.get("K8S_NAMESPACE", "autojanet")
 AGENT_IMAGE   = os.environ.get("AGENT_IMAGE", "registry.ctz.fyi/library/autojanet-agent:latest")
 
+# Max concurrent jobs per role (and global across all roles)
+MAX_JOBS_PER_ROLE   = int(os.environ.get("MAX_JOBS_PER_ROLE", "2"))
+MAX_JOBS_TOTAL      = int(os.environ.get("MAX_JOBS_TOTAL", "10"))
+MAX_TASK_RETRIES    = int(os.environ.get("MAX_TASK_RETRIES", "3"))
+
 # Standard bucket names (case-insensitive match)
 BUCKET_BACKLOG      = "backlog"
 BUCKET_TODO         = "todo"
 BUCKET_IN_PROGRESS  = "in progress"
-BUCKET_IN_REVIEW    = "in review"
+BUCKET_IN_REVIEW    = "review"
 BUCKET_DONE         = "done"
 
 VALID_ROLES = {
@@ -147,30 +156,49 @@ def vikunja_post(vikunja_token: str, path: str, body: dict) -> dict:
     return resp.json()
 
 
-def list_todo_tasks(vikunja_token: str, project_id: int, todo_id: int) -> list[dict]:
-    """Return all undone tasks in the Todo bucket with agent labels."""
+def list_tasks_in_bucket(vikunja_token: str, project_id: int, bucket_id: int) -> list[dict]:
+    """Return all undone tasks in a specific bucket using the filter API."""
     tasks = []
     page = 1
     while True:
-        batch = vikunja_get(vikunja_token, f"projects/{project_id}/tasks", page=page, per_page=50)
+        batch = vikunja_get(
+            vikunja_token,
+            f"projects/{project_id}/tasks",
+            page=page,
+            per_page=50,
+            filter=f"bucket_id = {bucket_id}",
+        )
         if not batch:
             break
         tasks.extend(batch)
         if len(batch) < 50:
             break
         page += 1
+    return [t for t in tasks if not t.get("done")]
+
+
+def list_todo_tasks(vikunja_token: str, project_id: int, todo_id: int) -> list[dict]:
+    """Return all undone tasks in the Todo bucket that have agent labels."""
     return [
-        t for t in tasks
-        if not t.get("done")
-        and t.get("labels")
-        and t.get("bucket_id") == todo_id
+        t for t in list_tasks_in_bucket(vikunja_token, project_id, todo_id)
+        if t.get("labels")
     ]
 
 
-def claim_task(vikunja_token: str, task_id: int, in_progress_id: int) -> bool:
-    """Move task from Todo → In Progress."""
+def claim_task(
+    vikunja_token: str,
+    task_id: int,
+    in_progress_id: int,
+    project_id: int,
+    view_id: int,
+) -> bool:
+    """Move task from Todo → In Progress via the kanban view endpoint."""
     try:
-        vikunja_post(vikunja_token, f"tasks/{task_id}", {"bucket_id": in_progress_id})
+        vikunja_post(
+            vikunja_token,
+            f"projects/{project_id}/views/{view_id}/buckets/{in_progress_id}/tasks",
+            {"task_id": task_id},
+        )
         log.info("Moved task %d → In Progress (bucket %d)", task_id, in_progress_id)
         return True
     except Exception as e:
@@ -178,10 +206,20 @@ def claim_task(vikunja_token: str, task_id: int, in_progress_id: int) -> bool:
         return False
 
 
-def unclaim_task(vikunja_token: str, task_id: int, todo_id: int) -> None:
-    """Move task back to Todo on job spawn failure."""
+def unclaim_task(
+    vikunja_token: str,
+    task_id: int,
+    todo_id: int,
+    project_id: int,
+    view_id: int,
+) -> None:
+    """Move task back to Todo on job spawn failure via the kanban view endpoint."""
     try:
-        vikunja_post(vikunja_token, f"tasks/{task_id}", {"bucket_id": todo_id})
+        vikunja_post(
+            vikunja_token,
+            f"projects/{project_id}/views/{view_id}/buckets/{todo_id}/tasks",
+            {"task_id": task_id},
+        )
         log.info("Unclaimed task %d → Todo", task_id)
     except Exception as e:
         log.warning("Failed to unclaim task %d: %s", task_id, e)
@@ -223,6 +261,251 @@ def sweep_done_tasks(
     log.info("Done sweep: moved %d tasks", moved)
 
 
+# ── Role extraction ───────────────────────────────────────────────────────────
+
+def extract_agent_role(task: dict) -> str | None:
+    """Return the agent role from an agent:<role> label, or None."""
+    for label in task.get("labels") or []:
+        title = label.get("title", "")
+        if title.startswith("agent:"):
+            role = title[len("agent:"):]
+            if role in VALID_ROLES:
+                return role
+    return None
+
+
+# ── Stale watchdog ────────────────────────────────────────────────────────────
+
+def count_jobs_for_task(batch_v1: k8s_client.BatchV1Api, task_id: int) -> tuple[int, int]:
+    """
+    Return (active_count, total_ever) for all jobs with this task_id label.
+    active = job has no terminal outcome yet (still running or pending).
+    total_ever = ALL jobs ever spawned for this task, regardless of outcome.
+    Used to detect both dead-job stale and successful-but-looping scenarios.
+    """
+    jobs = batch_v1.list_namespaced_job(
+        namespace=K8S_NAMESPACE,
+        label_selector=f"autojanet/task-id={task_id}",
+        _request_timeout=15,
+    )
+    active = 0
+    total_ever = len(jobs.items)
+    for job in jobs.items:
+        s = job.status
+        is_succeeded = (s.succeeded or 0) > 0
+        is_failed = (s.failed or 0) >= (job.spec.backoff_limit or 1) + 1
+        if not is_succeeded and not is_failed:
+            active += 1
+    return active, total_ever
+
+
+def post_stale_comment(vikunja_token: str, task_id: int, reason: str) -> None:
+    try:
+        httpx.put(
+            f"{VIKUNJA_BASE_URL}/api/v1/tasks/{task_id}/comments",
+            headers={"Authorization": f"Bearer {vikunja_token}", "Content-Type": "application/json"},
+            json={"comment": reason},
+            timeout=10,
+        ).raise_for_status()
+    except Exception as e:
+        log.warning("Failed to post stale comment on task %d: %s", task_id, e)
+
+
+def watchdog_stale_tasks(
+    vikunja_token: str,
+    batch_v1: k8s_client.BatchV1Api,
+    project_id: int,
+    view_id: int,
+    in_progress_id: int,
+    todo_id: int,
+    backlog_id: int,
+) -> None:
+    """
+    For every task stuck in InProgress, check whether its Job is still alive.
+    - No job found or job terminally failed:
+        - attempt count < MAX_TASK_RETRIES → move back to Todo
+        - attempt count >= MAX_TASK_RETRIES → move to Backlog + comment
+    """
+    stale = list_tasks_in_bucket(vikunja_token, project_id, in_progress_id)
+    log.info("Watchdog: checking %d InProgress tasks", len(stale))
+
+    for task in stale:
+        task_id = task["id"]
+        active, total_ever = count_jobs_for_task(batch_v1, task_id)
+
+        if active > 0:
+            log.debug("Task %d has %d active job(s), leaving alone", task_id, active)
+            continue
+
+        # No active job — task is stuck or looping
+        log.warning("Task %d is stale: active=%d total_attempts=%d", task_id, active, total_ever)
+
+        if total_ever >= MAX_TASK_RETRIES:
+            log.error("Task %d hit retry limit (%d/%d), moving to Backlog", task_id, total_ever, MAX_TASK_RETRIES)
+            try:
+                vikunja_post(
+                    vikunja_token,
+                    f"projects/{project_id}/views/{view_id}/buckets/{backlog_id}/tasks",
+                    {"task_id": task_id},
+                )
+            except Exception as e:
+                log.warning("Failed to move task %d to Backlog: %s", task_id, e)
+            post_stale_comment(
+                vikunja_token, task_id,
+                f"🚨 **Watchdog**: task has been attempted {total_ever} time(s) and hit the retry limit "
+                f"(`MAX_TASK_RETRIES={MAX_TASK_RETRIES}`). Moved to **Backlog** for manual review."
+            )
+        else:
+            log.info("Task %d attempt %d/%d, requeueing to Todo", task_id, total_ever + 1, MAX_TASK_RETRIES)
+            try:
+                vikunja_post(
+                    vikunja_token,
+                    f"projects/{project_id}/views/{view_id}/buckets/{todo_id}/tasks",
+                    {"task_id": task_id},
+                )
+            except Exception as e:
+                log.warning("Failed to requeue task %d to Todo: %s", task_id, e)
+            post_stale_comment(
+                vikunja_token, task_id,
+                f"⚠️ **Watchdog**: no active job found for this task (attempt {total_ever + 1}/{MAX_TASK_RETRIES}). "
+                f"Requeued to **Todo** for retry."
+            )
+
+
+# ── Review orchestration ──────────────────────────────────────────────────────
+
+# Roles that should NOT trigger review orchestration (would cause loops)
+REVIEW_SKIP_ROLES = {"pm", "code-reviewer"}
+
+
+def spawn_review_pm_job(
+    batch_v1: k8s_client.BatchV1Api,
+    task_id: int,
+    task_title: str,
+    in_review_bucket_id: int,
+    project_id: int,
+    view_id: int,
+) -> None:
+    """Spawn a PM agent job to orchestrate review sub-tasks for a completed task."""
+    name = f"review-pm-{task_id}"
+    if job_already_exists(batch_v1, name):
+        log.debug("Review PM job %s already exists, skipping", name)
+        return
+
+    job = k8s_client.V1Job(
+        api_version="batch/v1",
+        kind="Job",
+        metadata=k8s_client.V1ObjectMeta(
+            name=name,
+            namespace=K8S_NAMESPACE,
+            labels={
+                "autojanet/type": "review-pm",
+                "autojanet/role": "pm",
+                "autojanet/task-id": str(task_id),
+            },
+        ),
+        spec=k8s_client.V1JobSpec(
+            ttl_seconds_after_finished=3600,
+            backoff_limit=1,
+            template=k8s_client.V1PodTemplateSpec(
+                metadata=k8s_client.V1ObjectMeta(
+                    labels={
+                        "autojanet/type": "review-pm",
+                        "autojanet/role": "pm",
+                        "autojanet/task-id": str(task_id),
+                    }
+                ),
+                spec=k8s_client.V1PodSpec(
+                    service_account_name="agent-pm",
+                    restart_policy="Never",
+                    node_selector={"kubernetes.io/arch": "amd64"},
+                    containers=[
+                        k8s_client.V1Container(
+                            name="agent",
+                            image=AGENT_IMAGE,
+                            image_pull_policy="Always",
+                            env=[
+                                k8s_client.V1EnvVar(name="AGENT_ROLE",           value="pm"),
+                                k8s_client.V1EnvVar(name="TASK_TYPE",            value="review_orchestration"),
+                                k8s_client.V1EnvVar(name="TASK_ID",              value=str(task_id)),
+                                k8s_client.V1EnvVar(name="TASK_TITLE",           value=task_title),
+                                k8s_client.V1EnvVar(name="OPENBAO_ADDR",         value=OPENBAO_ADDR),
+                                k8s_client.V1EnvVar(name="VIKUNJA_BASE_URL",     value=VIKUNJA_BASE_URL),
+                                k8s_client.V1EnvVar(name="LITELLM_BASE_URL",     value="https://llm.ctz.fyi"),
+                                k8s_client.V1EnvVar(name="FORGEJO_BASE_URL",     value="https://git.ctz.fyi"),
+                                k8s_client.V1EnvVar(name="IN_REVIEW_BUCKET_ID",  value=str(in_review_bucket_id)),
+                                k8s_client.V1EnvVar(name="VIKUNJA_PROJECT_ID",   value=str(project_id)),
+                                k8s_client.V1EnvVar(name="VIKUNJA_VIEW_ID",      value=str(view_id)),
+                                k8s_client.V1EnvVar(
+                                    name="OPENBAO_ROLE_ID",
+                                    value_from=k8s_client.V1EnvVarSource(
+                                        secret_key_ref=k8s_client.V1SecretKeySelector(
+                                            name="agent-pm-approle", key="role_id",
+                                        )
+                                    ),
+                                ),
+                                k8s_client.V1EnvVar(
+                                    name="OPENBAO_SECRET_ID",
+                                    value_from=k8s_client.V1EnvVarSource(
+                                        secret_key_ref=k8s_client.V1SecretKeySelector(
+                                            name="agent-pm-approle", key="secret_id",
+                                        )
+                                    ),
+                                ),
+                            ],
+                            resources=k8s_client.V1ResourceRequirements(
+                                requests={"cpu": "250m", "memory": "512Mi"},
+                                limits={"cpu": "2000m", "memory": "2Gi"},
+                            ),
+                            security_context=k8s_client.V1SecurityContext(
+                                allow_privilege_escalation=False,
+                                run_as_non_root=True,
+                                run_as_user=1000,
+                                capabilities=k8s_client.V1Capabilities(drop=["ALL"]),
+                            ),
+                        )
+                    ],
+                ),
+            ),
+        ),
+    )
+
+    log.info("Spawning review PM job %s for task %d", name, task_id)
+    batch_v1.create_namespaced_job(namespace=K8S_NAMESPACE, body=job, _request_timeout=30)
+
+
+def orchestrate_review_tasks(
+    vikunja_token: str,
+    batch_v1: k8s_client.BatchV1Api,
+    project_id: int,
+    view_id: int,
+    in_review_id: int,
+) -> None:
+    """
+    Scan the Review bucket. For each task that has a non-pm/non-reviewer agent
+    label and no review-pm job yet, spawn a PM agent to create review sub-tasks.
+    """
+    review_tasks = list_tasks_in_bucket(vikunja_token, project_id, in_review_id)
+    log.info("Review orchestration: checking %d tasks in Review bucket", len(review_tasks))
+
+    for task in review_tasks:
+        task_id = task["id"]
+        role = extract_agent_role(task)
+
+        if not role or role in REVIEW_SKIP_ROLES:
+            log.info("Task %d role=%s skipped for review orchestration", task_id, role)
+            continue
+
+        spawn_review_pm_job(
+            batch_v1,
+            task_id=task_id,
+            task_title=task.get("title", f"Task {task_id}"),
+            in_review_bucket_id=in_review_id,
+            project_id=project_id,
+            view_id=view_id,
+        )
+
+
 # ── Kubernetes ────────────────────────────────────────────────────────────────
 
 def load_k8s_config() -> None:
@@ -232,6 +515,26 @@ def load_k8s_config() -> None:
         k8s_config.load_kube_config()
 
 
+def count_active_jobs(batch_v1: k8s_client.BatchV1Api) -> tuple[int, dict[str, int]]:
+    """Return (total_active, {role: count}) for all non-completed agent jobs."""
+    jobs = batch_v1.list_namespaced_job(
+        namespace=K8S_NAMESPACE,
+        label_selector="autojanet/type=agent",
+        _request_timeout=15,
+    )
+    total = 0
+    by_role: dict[str, int] = {}
+    for job in jobs.items:
+        # Skip completed/failed jobs
+        status = job.status
+        if (status.succeeded or 0) > 0 or (status.failed or 0) >= (job.spec.backoff_limit or 1) + 1:
+            continue
+        total += 1
+        role = job.metadata.labels.get("autojanet/role", "unknown")
+        by_role[role] = by_role.get(role, 0) + 1
+    return total, by_role
+
+
 def job_name(role: str, task_id: int) -> str:
     safe_role = role.replace("-", "")[:12]
     return f"agent-{safe_role}-{task_id}"
@@ -364,8 +667,9 @@ def main() -> None:
     in_progress_id  = buckets.get(BUCKET_IN_PROGRESS)
     in_review_id    = buckets.get(BUCKET_IN_REVIEW)
     done_id         = buckets.get(BUCKET_DONE)
+    backlog_id      = buckets.get(BUCKET_BACKLOG)
 
-    if not all([todo_id, in_progress_id, in_review_id, done_id]):
+    if not all([todo_id, in_progress_id, in_review_id, done_id, backlog_id]):
         log.error("Could not find all standard buckets. Found: %s", list(buckets.keys()))
         sys.exit(1)
 
@@ -373,13 +677,27 @@ def main() -> None:
     load_k8s_config()
     batch_v1 = k8s_client.BatchV1Api()
 
-    # Sweep: move any done=True tasks into the Done bucket
+    # 1. Sweep: move any done=True tasks into the Done bucket
     sweep_done_tasks(vikunja_token, project_id, view_id, done_id)
 
-    # Scan Todo bucket for claimable tasks
+    # 2. Watchdog: requeue or escalate stale InProgress tasks
+    watchdog_stale_tasks(
+        vikunja_token, batch_v1,
+        project_id, view_id,
+        in_progress_id, todo_id, backlog_id,
+    )
+
+    # 3. Review orchestration: spawn PM jobs for tasks awaiting review
+    orchestrate_review_tasks(vikunja_token, batch_v1, project_id, view_id, in_review_id)
+
+    # 4. Scan Todo bucket for claimable tasks
     tasks = list_todo_tasks(vikunja_token, project_id, todo_id)
     log.info("Found %d candidate tasks in Todo bucket", len(tasks))
 
+    # Check current job counts before spawning anything
+    total_active, active_by_role = count_active_jobs(batch_v1)
+    log.info("Active jobs: total=%d limits(per_role=%d, total=%d)", total_active, MAX_JOBS_PER_ROLE, MAX_JOBS_TOTAL)
+
     claimed = 0
     for task in tasks:
         task_id = task["id"]
@@ -390,16 +708,26 @@ def main() -> None:
             log.debug("Task %d has no valid agent label, skipping", task_id)
             continue
 
+        # Enforce concurrency limits
+        if total_active >= MAX_JOBS_TOTAL:
+            log.info("Global job limit reached (%d/%d), stopping", total_active, MAX_JOBS_TOTAL)
+            break
+        if active_by_role.get(role, 0) >= MAX_JOBS_PER_ROLE:
+            log.info("Role %s at limit (%d/%d), skipping task %d", role, active_by_role.get(role, 0), MAX_JOBS_PER_ROLE, task_id)
+            continue
+
         log.info("Claiming task %d (%s) for role=%s", task_id, title[:60], role)
-        if not claim_task(vikunja_token, task_id, in_progress_id):
+        if not claim_task(vikunja_token, task_id, in_progress_id, project_id, view_id):
             continue
 
         try:
             spawn_agent_job(batch_v1, role, task_id, title, in_review_id, project_id, view_id)
             claimed += 1
+            total_active += 1
+            active_by_role[role] = active_by_role.get(role, 0) + 1
         except Exception as e:
             log.error("Failed to spawn job for task %d: %s", task_id, e)
-            unclaim_task(vikunja_token, task_id, todo_id)
+            unclaim_task(vikunja_token, task_id, todo_id, project_id, view_id)
 
     log.info("Dispatcher done. Claimed %d tasks.", claimed)
 

intake/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY main.py .
+
+RUN useradd -m -u 1000 intake
+USER intake
+
+EXPOSE 8080
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

intake/main.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""
+AutoJanet Intake Service
+
+Accepts task submissions and creates Vikunja tasks with the appropriate
+agent label so the dispatcher picks them up automatically.
+
+POST /task
+  {
+    "title": "Add dark mode to the dashboard",
+    "description": "...",          # optional
+    "role": "coder"                # optional, defaults to "pm" (PM decomposes)
+  }
+
+GET /health
+"""
+
+import logging
+import os
+import sys
+
+import httpx
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout)
+log = logging.getLogger("intake")
+
+VIKUNJA_BASE_URL  = os.environ["VIKUNJA_BASE_URL"]
+VIKUNJA_PM_TOKEN  = os.environ["VIKUNJA_PM_TOKEN"]
+VIKUNJA_PROJECT_ID = int(os.environ.get("VIKUNJA_PROJECT_ID", "78"))
+VIKUNJA_TODO_BUCKET_ID = int(os.environ.get("VIKUNJA_TODO_BUCKET_ID", "116"))
+VIKUNJA_VIEW_ID   = int(os.environ.get("VIKUNJA_VIEW_ID", "114"))
+
+# Label IDs for agent roles (from Vikunja)
+ROLE_LABEL_IDS = {
+    "pm": 1,
+    "coder": 3,
+    "code-reviewer": 4,
+    "test-engineer": 5,
+    "devsecops": 6,
+    "secops": 7,
+    "sre": 8,
+    "kubernetes-pilot": 9,
+    "linux-admin": 10,
+    "systems-engineer": 11,
+    "networking": 12,
+    "dba": 13,
+    "prometheus-expert": 14,
+    "tofu-engineer": 15,
+    "release-manager": 16,
+    "doc-updater": 17,
+    "doc-writer": 18,
+    "technical-writer": 19,
+    "cost-optimizer": 20,
+}
+
+app = FastAPI(title="AutoJanet Intake", version="1.0.0")
+
+VIKUNJA_HEADERS = {
+    "Authorization": f"Bearer {VIKUNJA_PM_TOKEN}",
+    "Content-Type": "application/json",
+}
+
+
+class TaskRequest(BaseModel):
+    title: str
+    description: str = ""
+    role: str = "pm"
+
+
+class TaskResponse(BaseModel):
+    task_id: int
+    title: str
+    role: str
+    url: str
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok"}
+
+
+@app.post("/task", response_model=TaskResponse)
+def submit_task(req: TaskRequest):
+    role = req.role.lower()
+    if role not in ROLE_LABEL_IDS:
+        raise HTTPException(status_code=400, detail=f"Unknown role '{role}'. Valid: {sorted(ROLE_LABEL_IDS)}")
+
+    label_id = ROLE_LABEL_IDS[role]
+
+    # Create the task
+    with httpx.Client(timeout=15) as client:
+        resp = client.put(
+            f"{VIKUNJA_BASE_URL}/api/v1/projects/{VIKUNJA_PROJECT_ID}/tasks",
+            headers=VIKUNJA_HEADERS,
+            json={
+                "title": req.title,
+                "description": req.description,
+                "labels": [{"id": label_id}],
+            },
+        )
+        if resp.status_code not in (200, 201):
+            log.error("Vikunja task creation failed: %s %s", resp.status_code, resp.text)
+            raise HTTPException(status_code=502, detail="Failed to create Vikunja task")
+
+        task = resp.json()
+        task_id = task["id"]
+        log.info("Created task %d: %s (role=%s)", task_id, req.title, role)
+
+        # Move to Todo bucket
+        bucket_resp = client.post(
+            f"{VIKUNJA_BASE_URL}/api/v1/projects/{VIKUNJA_PROJECT_ID}/views/{VIKUNJA_VIEW_ID}/buckets/{VIKUNJA_TODO_BUCKET_ID}/tasks",
+            headers=VIKUNJA_HEADERS,
+            json={"task_id": task_id},
+        )
+        if bucket_resp.status_code not in (200, 201):
+            log.warning("Failed to move task %d to Todo bucket: %s", task_id, bucket_resp.text)
+
+        # Set the label explicitly (belt and suspenders)
+        label_resp = client.put(
+            f"{VIKUNJA_BASE_URL}/api/v1/tasks/{task_id}/labels",
+            headers=VIKUNJA_HEADERS,
+            json={"label_id": label_id},
+        )
+        if label_resp.status_code not in (200, 201):
+            log.warning("Failed to set label on task %d: %s", task_id, label_resp.text)
+
+    return TaskResponse(
+        task_id=task_id,
+        title=req.title,
+        role=role,
+        url=f"https://tasks.ctz.fyi/projects/{VIKUNJA_PROJECT_ID}/tasks/{task_id}",
+    )

intake/requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.6
+httpx==0.27.2
+pydantic==2.8.2

k8s/externalsecrets/intake-vikunja-token.yaml

@@ -0,0 +1,21 @@
+---
+# ExternalSecret: intake-vikunja-token
+# Pulls pm vikunja token from OpenBao for the intake service
+apiVersion: external-secrets.io/v1
+kind: ExternalSecret
+metadata:
+  name: intake-vikunja-token
+  namespace: autojanet
+spec:
+  refreshInterval: 1h
+  secretStoreRef:
+    name: openbao
+    kind: ClusterSecretStore
+  target:
+    name: intake-vikunja-token
+    creationPolicy: Owner
+  data:
+    - secretKey: token
+      remoteRef:
+        key: autojanet/pm/vikunja-token
+        property: token

k8s/manifests/dispatcher-cronjob.yaml

@@ -42,16 +42,14 @@ spec:
                       key: secret_id
                 - name: VIKUNJA_BASE_URL
                   value: "http://vikunja.vikunja.svc.cluster.local:3456"
-                - name: VIKUNJA_PROJECT_ID
-                  value: "78"
-                - name: VIKUNJA_TODO_BUCKET_ID
-                  value: "116"
-                - name: VIKUNJA_IN_PROGRESS_BUCKET_ID
-                  value: "117"
                 - name: K8S_NAMESPACE
                   value: "autojanet"
                 - name: AGENT_IMAGE
                   value: "registry.ctz.fyi/library/autojanet-agent:latest"
+                - name: MAX_JOBS_PER_ROLE
+                  value: "2"
+                - name: MAX_JOBS_TOTAL
+                  value: "10"
               resources:
                 requests:
                   cpu: "100m"

k8s/manifests/intake-deployment.yaml

@@ -0,0 +1,79 @@
+---
+# AutoJanet Intake Service
+# Accepts task submissions via HTTP and creates Vikunja tasks with agent labels.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: intake
+  namespace: autojanet
+  labels:
+    autojanet/role: intake
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      autojanet/role: intake
+  template:
+    metadata:
+      labels:
+        autojanet/role: intake
+    spec:
+      serviceAccountName: intake
+      containers:
+        - name: intake
+          image: registry.ctz.fyi/library/autojanet-intake:latest
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 8080
+          env:
+            - name: VIKUNJA_BASE_URL
+              value: "http://vikunja.vikunja.svc.cluster.local:3456"
+            - name: VIKUNJA_PROJECT_ID
+              value: "78"
+            - name: VIKUNJA_TODO_BUCKET_ID
+              value: "116"
+            - name: VIKUNJA_VIEW_ID
+              value: "114"
+            - name: VIKUNJA_PM_TOKEN
+              valueFrom:
+                secretKeyRef:
+                  name: intake-vikunja-token
+                  key: token
+          resources:
+            requests:
+              cpu: "50m"
+              memory: "64Mi"
+            limits:
+              cpu: "200m"
+              memory: "128Mi"
+          livenessProbe:
+            httpGet:
+              path: /health
+              port: 8080
+            initialDelaySeconds: 5
+            periodSeconds: 15
+          securityContext:
+            allowPrivilegeEscalation: false
+            runAsNonRoot: true
+            runAsUser: 1000
+            readOnlyRootFilesystem: true
+            capabilities:
+              drop: ["ALL"]
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: intake
+  namespace: autojanet
+spec:
+  selector:
+    autojanet/role: intake
+  ports:
+    - port: 80
+      targetPort: 8080
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: intake
+  namespace: autojanet

k8s/manifests/intake-ingress.yaml

@@ -0,0 +1,40 @@
+---
+# IngressRoute: janet.ctz.fyi → intake service
+apiVersion: traefik.io/v1alpha1
+kind: IngressRoute
+metadata:
+  name: intake
+  namespace: autojanet
+  annotations:
+    external-dns/internal: "true"
+    external-dns.alpha.kubernetes.io/hostname: janet.ctz.fyi
+spec:
+  entryPoints:
+    - websecure
+  routes:
+    - match: Host(`janet.ctz.fyi`)
+      kind: Rule
+      services:
+        - name: intake
+          port: 80
+  tls:
+    secretName: janet-ctz-fyi-tls
+---
+# Companion Ingress — cert-manager issues the cert, external-dns picks up the hostname
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: intake-dns
+  namespace: autojanet
+  annotations:
+    external-dns/internal: "true"
+    external-dns.alpha.kubernetes.io/hostname: janet.ctz.fyi
+    cert-manager.io/cluster-issuer: letsencrypt-prod
+spec:
+  ingressClassName: traefik
+  tls:
+    - hosts:
+        - janet.ctz.fyi
+      secretName: janet-ctz-fyi-tls
+  rules:
+    - host: janet.ctz.fyi