feat: enforce hard test-tier requirement across all DevPlace workflow agents and feature-builder docs

Update the feature-builder agent prompt, test-maintainer agent, and all four workflow JS files (devii-tool, endpoint, feature, job-service) to codify the DevPlace test standard as a non-optional project requirement: one test file per endpoint, directory tree mirroring the URL/source path, split into three tiers (unit, api, e2e). Add explicit Test phases to devii-tool, endpoint, feature, and job-service workflows, and embed tier-specific test instructions (path mapping, fixture choice, coverage scope) directly in each workflow's meta description and TESTS constant.

devplacepy/services/deepsearch/embeddings.py

@@ -6,7 +6,8 @@ import hashlib
 import logging
 import math
 import re
-from dataclasses import dataclass
+import time
+from dataclasses import dataclass, field
 
 from devplacepy import stealth
 from devplacepy.config import INTERNAL_EMBED_MODEL, INTERNAL_EMBED_URL
@@ -22,6 +23,23 @@ TOKEN_PATTERN = re.compile(r"[a-z0-9]+")
 class EmbedResult:
     vectors: list[list[float]]
     backend: str
+    latency_ms: int = 0
+    cache_hits: int = 0
+
+
+@dataclass
+class EmbeddingCache:
+    store: dict[str, list[float]] = field(default_factory=dict)
+
+    def key(self, text: str) -> str:
+        return hashlib.sha1((text or "").encode("utf-8")).hexdigest()
+
+    def get(self, text: str) -> list[float] | None:
+        return self.store.get(self.key(text))
+
+    def put(self, text: str, vector: list[float]) -> None:
+        if vector:
+            self.store[self.key(text)] = vector
 
 
 def _local_vector(text: str) -> list[float]:
@@ -41,19 +59,51 @@ def _local_vector(text: str) -> list[float]:
 
 
 def local_embed(texts: list[str]) -> EmbedResult:
-    return EmbedResult(vectors=[_local_vector(text) for text in texts], backend="local")
+    start = time.monotonic()
+    vectors = [_local_vector(text) for text in texts]
+    latency_ms = int((time.monotonic() - start) * 1000)
+    return EmbedResult(vectors=vectors, backend="local", latency_ms=latency_ms)
 
 
 async def embed_texts(
-    texts: list[str], api_key: str, *, gateway_url: str = INTERNAL_EMBED_URL
+    texts: list[str],
+    api_key: str,
+    *,
+    gateway_url: str = INTERNAL_EMBED_URL,
+    cache: EmbeddingCache | None = None,
 ) -> EmbedResult:
     if not texts:
         return EmbedResult(vectors=[], backend="empty")
+    cache_hits = 0
+    pending_index: list[int] = []
+    pending_text: list[str] = []
+    resolved: list[list[float] | None] = [None] * len(texts)
+    if cache is not None:
+        for index, text in enumerate(texts):
+            cached_vector = cache.get(text)
+            if cached_vector is not None:
+                resolved[index] = cached_vector
+                cache_hits += 1
+            else:
+                pending_index.append(index)
+                pending_text.append(text)
+    else:
+        pending_index = list(range(len(texts)))
+        pending_text = list(texts)
+    if not pending_text:
+        return EmbedResult(
+            vectors=[vector or [] for vector in resolved],
+            backend="cache",
+            latency_ms=0,
+            cache_hits=cache_hits,
+        )
     headers = {
         "Authorization": f"Bearer {api_key}",
         "Content-Type": "application/json",
     }
-    payload = {"model": INTERNAL_EMBED_MODEL, "input": texts}
+    payload = {"model": INTERNAL_EMBED_MODEL, "input": pending_text}
+    start = time.monotonic()
+    backend = "gateway"
     try:
         async with stealth.stealth_async_client(timeout=EMBED_TIMEOUT_SECONDS) as client:
             response = await client.post(gateway_url, json=payload, headers=headers)
@@ -61,10 +111,22 @@ async def embed_texts(
             raise RuntimeError(f"embed gateway returned {response.status_code}")
         data = response.json()
         rows = data.get("data") or []
-        vectors = [row.get("embedding") or [] for row in rows]
-        if len(vectors) != len(texts) or any(not vector for vector in vectors):
+        fresh = [row.get("embedding") or [] for row in rows]
+        if len(fresh) != len(pending_text) or any(not vector for vector in fresh):
             raise RuntimeError("embed gateway returned an incomplete response")
-        return EmbedResult(vectors=vectors, backend="gateway")
     except Exception as exc:
         logger.warning("deepsearch embedding gateway failed, using local: %s", exc)
-        return local_embed(texts)
+        fresh = [_local_vector(text) for text in pending_text]
+        backend = "local"
+    latency_ms = int((time.monotonic() - start) * 1000)
+    for offset, index in enumerate(pending_index):
+        vector = fresh[offset]
+        resolved[index] = vector
+        if cache is not None:
+            cache.put(pending_text[offset], vector)
+    return EmbedResult(
+        vectors=[vector or [] for vector in resolved],
+        backend=backend,
+        latency_ms=latency_ms,
+        cache_hits=cache_hits,
+    )

devplacepy/services/deepsearch/store.py

@@ -15,8 +15,7 @@ logger = logging.getLogger(__name__)
 TOKEN_PATTERN = re.compile(r"[a-z0-9]+")
 BM25_K1 = 1.5
 BM25_B = 0.75
-HYBRID_VECTOR_WEIGHT = 0.6
-HYBRID_KEYWORD_WEIGHT = 0.4
+RRF_K = 60.0
 DEFAULT_TOP_K = 8
 CANDIDATE_MULTIPLIER = 4
 
@@ -182,18 +181,39 @@ class VectorStore:
         )
         if not candidates:
             return []
+        vector_rank = sorted(
+            candidates, key=lambda chunk: chunk.score, reverse=True
+        )
         keyword = self.keyword_scores(query, candidates)
-        vec_max = max((chunk.score for chunk in candidates), default=0.0) or 1.0
-        kw_max = max(keyword.values(), default=0.0) or 1.0
+        keyword_rank = sorted(
+            candidates, key=lambda chunk: keyword.get(chunk.uid, 0.0), reverse=True
+        )
+        fused: dict[str, float] = {}
+        for rank, chunk in enumerate(vector_rank, start=1):
+            fused[chunk.uid] = fused.get(chunk.uid, 0.0) + 1.0 / (RRF_K + rank)
+        for rank, chunk in enumerate(keyword_rank, start=1):
+            fused[chunk.uid] = fused.get(chunk.uid, 0.0) + 1.0 / (RRF_K + rank)
         for chunk in candidates:
-            vec_norm = max(0.0, chunk.score) / vec_max
-            kw_norm = keyword.get(chunk.uid, 0.0) / kw_max
-            chunk.score = (
-                HYBRID_VECTOR_WEIGHT * vec_norm + HYBRID_KEYWORD_WEIGHT * kw_norm
-            )
+            chunk.score = fused.get(chunk.uid, 0.0)
         candidates.sort(key=lambda chunk: chunk.score, reverse=True)
         return candidates[:top_k]
 
+    def coverage_analytics(self) -> dict:
+        chunks = self.all_chunks()
+        if not chunks:
+            return {"chunks": 0, "domains": 0, "sources": 0, "avg_chunk_chars": 0}
+        domains = {chunk.metadata.get("url", "") for chunk in chunks}
+        domains.discard("")
+        sources = {chunk.source for chunk in chunks}
+        sources.discard("")
+        avg_chars = int(sum(len(chunk.text) for chunk in chunks) / len(chunks))
+        return {
+            "chunks": len(chunks),
+            "domains": len(domains),
+            "sources": len(sources),
+            "avg_chunk_chars": avg_chars,
+        }
+
     def drop(self) -> None:
         try:
             import chromadb

devplacepy/services/devii/actions/catalog.py

@@ -197,6 +197,16 @@ ACTIONS: tuple[Action, ...] = (
             body("attachment_uids", ATTACHMENTS),
         ),
     ),
+    Action(
+        name="edit_comment",
+        method="POST",
+        path="/comments/edit/{comment_uid}",
+        summary="Edit the body of one of your own comments",
+        params=(
+            path("comment_uid", "Uid of the comment."),
+            body("content", "New comment body, 3-1000 characters.", required=True),
+        ),
+    ),
     Action(
         name="delete_comment",
         method="POST",
@@ -910,6 +920,20 @@ ACTIONS: tuple[Action, ...] = (
             body("description", "Issue description.", required=True),
         ),
     ),
+    Action(
+        name="planning_report_generate",
+        method="POST",
+        path="/issues/planning",
+        summary="Queue an admin planning report of all open tickets.",
+        description=(
+            "Queues a background job that builds a grouped, ordered markdown planning report "
+            "of every open ticket and returns {uid, status_url}. Poll the status_url until "
+            "status is 'done', then show the markdown and the download_url. Admin only."
+        ),
+        params=(),
+        requires_auth=True,
+        requires_admin=True,
+    ),
     Action(
         name="issue_job_status",
         method="GET",

devplacepy/services/gitea/planning.py

@@ -0,0 +1,159 @@
+# retoor <retoor@molodetz.nl>
+
+import logging
+
+import httpx
+
+from devplacepy import stealth
+from devplacepy.config import INTERNAL_GATEWAY_URL
+from devplacepy.services.gitea.config import HTTP_TIMEOUT_SECONDS, GiteaConfig
+
+logger = logging.getLogger(__name__)
+
+MAX_TOKENS = 3000
+MAX_ISSUES = 50
+BODY_EXCERPT_MAX = 600
+PLAN_MAX = 80000
+UNGROUPED_LABEL = "General"
+
+SYSTEM_PROMPT = (
+    "You are a senior engineering lead for DevPlace, a social network for software "
+    "developers. You are given the full list of OPEN issue tickets from a Gitea tracker. "
+    "Produce a single, clear delivery plan in GitHub-flavored markdown that GROUPS the "
+    "tickets into logical workstreams and ORDERS them so dependencies and high-impact work "
+    "come first. Never invent tickets, numbers, or facts that are not in the input.\n\n"
+    "Return ONLY markdown, using EXACTLY this structure:\n"
+    "# Open Tickets Planning\n"
+    "A short paragraph summarising the overall plan and how many tickets it covers.\n\n"
+    "Then one level-2 heading per group (a workstream or theme). Under each group, an "
+    "ordered (numbered) list of its tickets. Each list item MUST start with the ticket "
+    "number as '#N', then the title, then a single sentence explaining the chosen order "
+    "or the work to do, for example:\n"
+    "1. #12 Fix the login redirect - blocks every authenticated flow, do first.\n\n"
+    "End with a level-2 heading '## Suggested Order' giving a flat numbered list of every "
+    "ticket number in the recommended execution order."
+)
+
+
+def _excerpt(text: str) -> str:
+    text = (text or "").strip().replace("\r\n", "\n")
+    if len(text) > BODY_EXCERPT_MAX:
+        return text[:BODY_EXCERPT_MAX].rstrip() + "..."
+    return text
+
+
+def _labels(issue: dict) -> list[str]:
+    labels = issue.get("labels") or []
+    names: list[str] = []
+    for label in labels:
+        if isinstance(label, dict):
+            name = str(label.get("name", "")).strip()
+        else:
+            name = str(label).strip()
+        if name:
+            names.append(name)
+    return names
+
+
+def _group_key(issue: dict) -> str:
+    labels = _labels(issue)
+    return labels[0] if labels else UNGROUPED_LABEL
+
+
+def _fallback(issues: list[dict]) -> str:
+    groups: dict[str, list[dict]] = {}
+    for issue in issues:
+        groups.setdefault(_group_key(issue), []).append(issue)
+
+    ordered_group_names = sorted(
+        groups,
+        key=lambda name: (name == UNGROUPED_LABEL, name.lower()),
+    )
+
+    lines: list[str] = ["# Open Tickets Planning", ""]
+    lines.append(
+        f"Deterministic plan covering {len(issues)} open ticket"
+        f"{'' if len(issues) == 1 else 's'}, grouped by primary label and ordered by "
+        "ticket number within each group."
+    )
+    lines.append("")
+
+    flat_order: list[int] = []
+    for name in ordered_group_names:
+        members = sorted(groups[name], key=lambda i: int(i.get("number", 0)))
+        lines.append(f"## {name}")
+        for position, issue in enumerate(members, start=1):
+            number = int(issue.get("number", 0))
+            title = str(issue.get("title", "")).strip() or "Untitled"
+            excerpt = _excerpt(issue.get("body", ""))
+            rationale = excerpt or "No description provided."
+            lines.append(f"{position}. #{number} {title} - {rationale}")
+            flat_order.append(number)
+        lines.append("")
+
+    lines.append("## Suggested Order")
+    for position, number in enumerate(flat_order, start=1):
+        lines.append(f"{position}. #{number}")
+    lines.append("")
+
+    return "\n".join(lines)[:PLAN_MAX]
+
+
+def _user_message(issues: list[dict]) -> str:
+    parts: list[str] = ["Open tickets to plan:", ""]
+    for issue in issues:
+        number = int(issue.get("number", 0))
+        title = str(issue.get("title", "")).strip() or "Untitled"
+        labels = ", ".join(_labels(issue)) or "none"
+        excerpt = _excerpt(issue.get("body", ""))
+        parts.append(f"#{number} {title}")
+        parts.append(f"labels: {labels}")
+        parts.append(f"description: {excerpt or 'none'}")
+        parts.append("")
+    return "\n".join(parts)
+
+
+async def generate_plan(
+    issues: list[dict], config: GiteaConfig
+) -> tuple[str, bool]:
+    issues = list(issues)[:MAX_ISSUES]
+    if not issues:
+        return "# Open Tickets Planning\n\nThere are no open tickets to plan.\n", False
+    if not config.ai_enhance:
+        return _fallback(issues), False
+
+    payload = {
+        "model": config.ai_model,
+        "messages": [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": _user_message(issues)},
+        ],
+        "max_tokens": MAX_TOKENS,
+        "temperature": 0.2,
+    }
+    headers = {"Content-Type": "application/json"}
+    if config.ai_key:
+        headers["Authorization"] = f"Bearer {config.ai_key}"
+
+    try:
+        async with stealth.stealth_async_client(timeout=HTTP_TIMEOUT_SECONDS) as client:
+            response = await client.post(
+                INTERNAL_GATEWAY_URL, json=payload, headers=headers
+            )
+            response.raise_for_status()
+            content = (
+                response.json()
+                .get("choices", [{}])[0]
+                .get("message", {})
+                .get("content", "")
+            )
+    except (httpx.HTTPError, ValueError, KeyError, IndexError) as exc:
+        logger.warning("Planning generation failed, using fallback: %s", exc)
+        return _fallback(issues), False
+
+    markdown = (content or "").strip()
+    if not markdown:
+        logger.warning("Planning generation returned empty content, using fallback")
+        return _fallback(issues), False
+    logger.info("Generated AI planning for %d open tickets", len(issues))
+    return markdown[:PLAN_MAX], True

devplacepy/services/jobs/deepsearch/crawl.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import logging
 import re
+import time
 from dataclasses import dataclass, field
 from typing import Awaitable, Callable
 
@@ -14,6 +16,7 @@ from devplacepy import stealth
 from devplacepy.net_guard import BlockedAddressError, guard_public_url, guarded_async_client
 
 from .pdf import MAX_PDF_BYTES, extract_pdf_text, is_pdf
+from .phases import PHASE_CRAWLING
 
 logger = logging.getLogger(__name__)
 
@@ -22,6 +25,7 @@ RSEARCH_TIMEOUT_SECONDS = 45.0
 FETCH_TIMEOUT_SECONDS = 20.0
 MAX_FETCH_BYTES = 2_500_000
 RESULTS_PER_QUERY = 8
+CRAWL_CONCURRENCY = 4
 USER_AGENT = (
     "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
     "Chrome/131.0.0.0 Safari/537.36 DevPlaceDeepSearchBot/1.0"

devplacepy/services/jobs/deepsearch/enhance.py

@@ -5,10 +5,13 @@ from __future__ import annotations
 import json
 import logging
 import re
+from typing import Callable
 
 from devplacepy import stealth
 from devplacepy.config import INTERNAL_GATEWAY_URL, INTERNAL_MODEL
 
+from .phases import PHASE_PLANNING
+
 logger = logging.getLogger(__name__)
 
 ENHANCE_TIMEOUT_SECONDS = 90.0
@@ -52,7 +55,20 @@ def _parse(text: str) -> list[str]:
     return cleaned[:MAX_SUBQUERIES]
 
 
-async def plan_queries(query: str, api_key: str) -> list[str]:
+def _noop(frame: dict) -> None:
+    return None
+
+
+async def plan_queries(
+    query: str, api_key: str, emit: Callable[[dict], None] = _noop
+) -> list[str]:
+    emit(
+        {
+            "type": "substep",
+            "phase": PHASE_PLANNING,
+            "message": "Drafting diverse search angles",
+        }
+    )
     payload = {
         "model": INTERNAL_MODEL,
         "messages": [
@@ -79,7 +95,22 @@ async def plan_queries(query: str, api_key: str) -> list[str]:
         )
         parsed = _parse(content)
         if parsed:
+            emit(
+                {
+                    "type": "substep",
+                    "phase": PHASE_PLANNING,
+                    "message": f"Planned {len(parsed)} search angles",
+                }
+            )
             return parsed
     except Exception as exc:
         logger.warning("deepsearch query planner failed, using fallback: %s", exc)
-    return _fallback(query)
+    fallback = _fallback(query)
+    emit(
+        {
+            "type": "substep",
+            "phase": PHASE_PLANNING,
+            "message": f"Planner unavailable, using {len(fallback)} heuristic angles",
+        }
+    )
+    return fallback

devplacepy/services/jobs/deepsearch/phases.py

@@ -0,0 +1,38 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+PHASE_PLANNING = "planning"
+PHASE_SEARCHING = "searching"
+PHASE_CRAWLING = "crawling"
+PHASE_INDEXING = "indexing"
+PHASE_ANALYSIS = "analysis"
+PHASE_SYNTHESIS = "synthesis"
+
+PHASE_ORDER: list[str] = [
+    PHASE_PLANNING,
+    PHASE_SEARCHING,
+    PHASE_CRAWLING,
+    PHASE_INDEXING,
+    PHASE_ANALYSIS,
+    PHASE_SYNTHESIS,
+]
+
+PHASE_LABELS: dict[str, str] = {
+    PHASE_PLANNING: "Planning",
+    PHASE_SEARCHING: "Searching",
+    PHASE_CRAWLING: "Crawling",
+    PHASE_INDEXING: "Indexing",
+    PHASE_ANALYSIS: "Analysis",
+    PHASE_SYNTHESIS: "Synthesis",
+}
+
+TOTAL_PHASES = len(PHASE_ORDER)
+HEARTBEAT_SECONDS = 2.0
+
+
+def phase_index(phase: str) -> int:
+    try:
+        return PHASE_ORDER.index(phase)
+    except ValueError:
+        return 0

devplacepy/services/jobs/planning_service.py

@@ -0,0 +1,119 @@
+# retoor <retoor@molodetz.nl>
+
+import logging
+import zlib
+from pathlib import Path
+
+from devplacepy.attachments import _directory_for
+from devplacepy.config import PLANNING_REPORTS_DIR
+from devplacepy.services.gitea import runtime
+from devplacepy.services.gitea.client import STATE_OPEN, GiteaError
+from devplacepy.services.gitea.config import gitea_config
+from devplacepy.services.gitea.planning import MAX_ISSUES, generate_plan
+from devplacepy.services.jobs.base import JobService
+from devplacepy.utils import slugify
+
+logger = logging.getLogger(__name__)
+
+PAGE_LIMIT = 50
+
+
+class PlanningReportService(JobService):
+    kind = "planning"
+    title = "Ticket planning"
+    description = (
+        "Builds a grouped, ordered markdown planning report of all open Gitea tickets off "
+        "the request path using the internal AI service (with a deterministic fallback), "
+        "writes the markdown artifact, and prunes it once unused for the retention window."
+    )
+
+    def __init__(self) -> None:
+        super().__init__(name="planning", interval_seconds=2)
+
+    async def process(self, job: dict) -> dict:
+        from devplacepy.services.audit import record as audit
+
+        uid: str = job["uid"]
+        owner_kind: str = job.get("owner_kind") or "system"
+        owner_id: str = job.get("owner_id") or ""
+        try:
+            config = gitea_config()
+            if not config.is_configured:
+                raise GiteaError("Gitea integration is not configured", status=503)
+
+            issues = await self._collect_open(config)
+            markdown, ai_used = await generate_plan(issues, config)
+            issue_count = len(issues)
+
+            final_name = self._final_name(job.get("preferred_name", ""), markdown)
+            target_dir = PLANNING_REPORTS_DIR / _directory_for(uid)
+            target_dir.mkdir(parents=True, exist_ok=True)
+            final_path = target_dir / final_name
+            final_path.write_text(markdown, encoding="utf-8")
+            bytes_out = len(markdown.encode("utf-8"))
+        except Exception:
+            audit.record_system(
+                "issue.planning.generate",
+                actor_kind="user" if owner_kind == "user" else owner_kind,
+                actor_uid=owner_id if owner_kind == "user" else None,
+                result="failure",
+                target_type="issue",
+                metadata={"job": uid},
+                summary="planning report generation failed",
+                links=[audit.job(uid)],
+            )
+            raise
+
+        audit.record_system(
+            "issue.planning.generate",
+            actor_kind="user" if owner_kind == "user" else owner_kind,
+            actor_uid=owner_id if owner_kind == "user" else None,
+            target_type="issue",
+            metadata={
+                "issue_count": issue_count,
+                "ai_used": ai_used,
+                "bytes_out": bytes_out,
+            },
+            summary=f"planning report for {issue_count} open tickets generated",
+            links=[audit.job(uid)],
+        )
+        self.log(f"Planning job {uid} covered {issue_count} open tickets")
+        return {
+            "download_url": f"/issues/planning/{uid}/download",
+            "local_path": str(final_path),
+            "final_name": final_name,
+            "markdown": markdown,
+            "ai_used": ai_used,
+            "item_count": issue_count,
+            "bytes_out": bytes_out,
+        }
+
+    async def _collect_open(self, config) -> list[dict]:
+        client = runtime.get_client()
+        collected: list[dict] = []
+        page = 1
+        while len(collected) < MAX_ISSUES:
+            issues, total = await client.list_issues(
+                state=STATE_OPEN, page=page, limit=PAGE_LIMIT
+            )
+            if not issues:
+                break
+            collected.extend(issues)
+            if len(issues) < PAGE_LIMIT or len(collected) >= total:
+                break
+            page += 1
+        return collected[:MAX_ISSUES]
+
+    def _final_name(self, preferred_name: str, markdown: str) -> str:
+        name = preferred_name or "open-tickets-plan"
+        if name.lower().endswith(".md"):
+            name = name[:-3]
+        slug = slugify(name) or "open-tickets-plan"
+        crc32 = zlib.crc32(markdown.encode("utf-8")) & 0xFFFFFFFF
+        return f"{crc32:08x}.{slug}.md"
+
+    def cleanup(self, job: dict) -> None:
+        result = job.get("result", {})
+        local_path = result.get("local_path")
+        if local_path:
+            Path(local_path).unlink(missing_ok=True)