diff --git a/.claude/skills/hugagent-frontend-dev/SKILL.md b/.claude/skills/hugagent-frontend-dev/SKILL.md
index d0743241..5f2c624b 100644
--- a/.claude/skills/hugagent-frontend-dev/SKILL.md
+++ b/.claude/skills/hugagent-frontend-dev/SKILL.md
@@ -124,7 +124,7 @@ src/frontend/src/
│ ├── variables.css # CSS 自定义属性(颜色、阴影)
│ ├── common.css # 全局工具类
│ ├── chat.css / sidebar.css / catalog.css / tool.css
-│ ├── automation.css / automation-timeline.css / batch(并入相关文件)
+│ ├── automation.css / batch(并入相关文件)
│ ├── canvas.css / config.css / mcp.css / myspace.css / plan.css
│ └── projects.css / search-modal.css / settings.css / skill-distill.css / team-folder.css
```
diff --git a/document/en/modules/automation.md b/document/en/modules/automation.md
index d2c375f6..1f88aa8a 100644
--- a/document/en/modules/automation.md
+++ b/document/en/modules/automation.md
@@ -58,7 +58,7 @@ Scheduled prompts and ordinary conversations share admission, capability resolut
Scheduled Tasks is a dedicated module under src/frontend/src/components/automation/. Its sidebar groups conversations by task ID and shows one entry per scheduled task, keeping distinct tasks separate even when their names match. New executions update the entry's latest activity time; clicking it opens the newest execution. The right control panel retains individual executions, and choosing a historical run keeps the same task selected in the sidebar. Setup conversations remain separate, and empty setup conversations are hidden. The oldest conversation anchors each entry, preserving its ordering, pin, favorite, and rename actions. Export and delete apply to the selected execution, or the newest execution when the task is not open; deleting one conversation does not delete the scheduled task. All Tasks opens configuration, status, and run history in the content area. Creating through chat first persists a conversation owned by Scheduled Tasks, references the installed task-management plugin, and prefills an editable template. Task conversations are excluded from the ordinary chat sidebar.
-Execution conversations use /automation/:taskId/conversations/:chatId, with new as the task segment for setup conversations. Reloads, browser history, search hits, and notifications preserve task ownership. Search opens the exact matching conversation instead of the latest run. RunTimelinePanel groups executions by date. automationStore owns the single task list; automationChatStore owns the current run group and task pin/favorite preferences.
+Execution conversations use /automation/:taskId/conversations/:chatId, with new as the task segment for setup conversations. Reloads, browser history, search hits, and notifications preserve task ownership. Search opens the exact matching conversation instead of the latest run. The run monitor rail beside an execution conversation shows a scheduled-task card listing every run, newest first. automationStore owns the single task list; automationChatStore owns the current run group and task pin/favorite preferences.
Opening the right-side Canvas in an execution conversation temporarily hides the run timeline to make room for the conversation and Canvas. Collapsing Canvas restores the timeline and preserves the current task, selected execution, and conversation.
@@ -106,7 +106,7 @@ Every morning an Industry morning brief execution conversation is created. The s
| Plan REST API | `src/backend/api/routes/v1/plans.py` |
| Plan generation / execution orchestration | `src/backend/orchestration/plans/`, `src/backend/core/chat/plans.py` |
| Plan / step models | `src/backend/core/db/models/agent.py` |
-| Frontend automation UI | `src/frontend/src/components/automation/AutomationPanel.tsx` etc., `src/frontend/src/components/automation/RunTimelinePanel.tsx` |
+| Frontend automation UI | `src/frontend/src/components/automation/AutomationPanel.tsx` etc., `src/frontend/src/components/monitor/ScheduledTaskCard.tsx` |
| Frontend state | `src/frontend/src/stores/automationStore.ts`, `automationChatStore.ts` |
Further reading: [Chat System](chat.md) · [MCP Tools](mcp-tools.md) · [Canvas & Artifacts](canvas-artifacts.md)
diff --git a/document/en/modules/projects-myspace.md b/document/en/modules/projects-myspace.md
index ad5faf7d..7fd1cdbc 100644
--- a/document/en/modules/projects-myspace.md
+++ b/document/en/modules/projects-myspace.md
@@ -167,7 +167,7 @@ Permission model: `TeamMember.role` (owner/admin/member) + `file_permission` (vi
Three complementary paths:
-1. **Attachment injection (hooks)**: frontend requests send only `file_id`, `name`, and `mime_type` in each `attachments[]` item, never the file body. `core/harness/hooks.py` lazily downloads and parses by `file_id`, caching `parsed_text` and its summary; the first turn injects only metadata and a bounded preview of at most 5,000 characters, while xlsx previews also include total dimensions, headers, and a few data rows. The model reads content beyond the preview on demand through `read_artifact`; images still use multimodal injection. Every read validates `user_id` ownership to block forged cross-user access;
+1. **Attachment injection (hooks)**: frontend requests send only `file_id`, `name`, and `mime_type` in each `attachments[]` item, never the file body. The attachment hook (`core/llm/context/attachment_hook.py`) neither downloads nor parses files: it injects only each file's name, MIME type, and `file_id` plus reading instructions, so attachments add nothing to time-to-first-token. When the model needs the content it calls `read_artifact`, which downloads and parses by `file_id`, caches `parsed_text` and its summary, and returns paginated text (xlsx by sheet, pptx by slide); images still use multimodal injection. Every read validates `user_id` ownership to block forged cross-user access;
2. **Project file access**: a project conversation's system prompt supplies project instructions and directory scope without an automatic file inventory or count. The agent discovers files on demand through directory, search and read tools;
3. **Sandbox virtual filesystem**: with code execution enabled, the `/myspace/...` path maps MySpace into the sandbox (lazy loading + reverse sync); team projects map the team folder — see [Sandbox](./sandbox.md).
diff --git a/document/zh-CN/modules/automation.md b/document/zh-CN/modules/automation.md
index 2c53f631..dd4f24ea 100644
--- a/document/zh-CN/modules/automation.md
+++ b/document/zh-CN/modules/automation.md
@@ -58,7 +58,7 @@ HugAgentOS 内置两种「把一句话变成可重复的生产力」机制,两
定时任务是独立模块,相关组件位于 src/frontend/src/components/automation/。左侧按任务 ID 合并会话,每个定时任务只显示一条记录;同名的不同任务分别显示。新运行更新入口的最近活动时间,点击入口打开最近一次运行。右侧控制面板保留各次执行记录,选择历史运行后左侧仍选中同一个任务。创建中的会话分别显示,空白创建会话隐藏。侧边栏沿用最早会话作为稳定入口,保留该会话的排序、置顶、收藏和重命名操作。导出和删除作用于当前选中的执行会话,未打开该任务时作用于最近一次执行;删除单次会话不删除定时任务。点击“全部任务”进入内容区管理任务配置、状态和执行记录。通过对话创建时,会先保存归属于定时任务模块的会话,引用已安装的定时任务管理插件并填入可编辑模板。普通对话列表不展示任务会话。
-执行会话使用 /automation/:taskId/conversations/:chatId;创建会话使用任务段 new。刷新、前进后退、搜索命中及通知入口均保留任务归属;搜索执行结果直接打开命中的会话,避免误跳到最新执行。RunTimelinePanel 提供按日期分组的执行记录。automationStore 管理唯一任务列表,automationChatStore 仅管理当前执行组和任务置顶、收藏偏好。
+执行会话使用 /automation/:taskId/conversations/:chatId;创建会话使用任务段 new。刷新、前进后退、搜索命中及通知入口均保留任务归属;搜索执行结果直接打开命中的会话,避免误跳到最新执行。执行会话右侧的运行监控栏提供「定时任务」卡片,按时间倒序列出全部执行记录。automationStore 管理唯一任务列表,automationChatStore 仅管理当前执行组和任务置顶、收藏偏好。
执行会话打开右侧 Canvas 时,执行记录面板暂时隐藏,为对话和 Canvas 留出空间。收起 Canvas 后恢复执行记录面板,保留当前任务、选中的执行和会话。
@@ -106,7 +106,7 @@ POST /v1/automations
| 计划 REST API | `src/backend/api/routes/v1/plans.py` |
| 计划生成 / 执行编排 | `src/backend/orchestration/plans/`、`src/backend/core/chat/plans.py` |
| 计划 / 步骤模型 | `src/backend/core/db/models/agent.py` |
-| 前端自动化 UI | `src/frontend/src/components/automation/AutomationPanel.tsx` 等、`src/frontend/src/components/automation/RunTimelinePanel.tsx` |
+| 前端自动化 UI | `src/frontend/src/components/automation/AutomationPanel.tsx` 等、`src/frontend/src/components/monitor/ScheduledTaskCard.tsx` |
| 前端状态 | `src/frontend/src/stores/automationStore.ts`、`automationChatStore.ts` |
延伸阅读:[对话系统](chat.md) · [MCP 工具](mcp-tools.md) · [数据画布与产物](canvas-artifacts.md)
diff --git a/document/zh-CN/modules/projects-myspace.md b/document/zh-CN/modules/projects-myspace.md
index 7f009df6..a52b990c 100644
--- a/document/zh-CN/modules/projects-myspace.md
+++ b/document/zh-CN/modules/projects-myspace.md
@@ -158,7 +158,7 @@ teams ─────────┬── team_members(role: owner/admin/memb
三条路径,互为补充:
-1. **附件注入(hooks)**:前端请求的 `attachments[]` 只携带 `file_id`、`name`、`mime_type`,不携带文件正文。`core/harness/hooks.py` 按 `file_id` 延迟下载、解析并缓存 `parsed_text` 与摘要;首轮只注入文件元数据和最多 5,000 字符的有界预览,xlsx 额外提供总行列数、表头和少量数据行。模型需要预览外的正文时通过 `read_artifact` 按需读取;图片仍走多模态注入。所有读取都校验 `user_id` 归属,防止伪造 `file_id` 跨用户访问;
+1. **附件注入(hooks)**:前端请求的 `attachments[]` 只携带 `file_id`、`name`、`mime_type`,不携带文件正文。附件钩子(`core/llm/context/attachment_hook.py`)不下载也不解析文件,只注入文件名、MIME 和 `file_id` 以及读取指引,因此附件不会拖慢首字响应。模型需要正文时调用 `read_artifact`,由它按 `file_id` 下载、解析并缓存 `parsed_text` 与摘要,再分页返回(xlsx 可按 sheet、pptx 可按页读取);图片仍走多模态注入。所有读取都校验 `user_id` 归属,防止伪造 `file_id` 跨用户访问;
2. **项目文件访问**:项目对话的 system prompt 提供项目指令和目录范围,不自动携带文件列表或文件总数。agent 按需使用目录、搜索和读取工具获取文件信息;
3. **沙箱虚拟文件系统**:开启代码执行时,`/myspace/...` 路径把我的空间映射进沙箱(懒加载 + 文件系统实时登记),团队项目映射团队文件夹——见 [沙箱](./sandbox.md)。
diff --git a/scripts/refactor/import_contract.py b/scripts/refactor/import_contract.py
index abf1c1ca..ab576048 100644
--- a/scripts/refactor/import_contract.py
+++ b/scripts/refactor/import_contract.py
@@ -28,6 +28,7 @@
("automation", "desktop"), # Authorized local task dispatch; see module README.
("distillation", "marketplace"), # Explicit draft publication; see module README.
("security_console", "sandbox_ops"), # Read-only rebuild status; see module README.
+ ("task_monitor", "security_console"), # Read-only health snapshot; see module README.
}
OUTER = ("features", "integrations", "personal_agent", "orchestration", "api")
diff --git a/src/backend/alembic/versions/ce_0036_task_numbers.py b/src/backend/alembic/versions/ce_0036_task_numbers.py
new file mode 100644
index 00000000..e6d9f8cf
--- /dev/null
+++ b/src/backend/alembic/versions/ce_0036_task_numbers.py
@@ -0,0 +1,28 @@
+"""Task numbers: RW-YYYYMMDD-NNNN on every run, RW-…-NN on every sub-agent call.
+
+Same schema and backfill as the EE revision ``tasknumber01`` (``core.db.task_number_migration``).
+
+Revision ID: ce_0036
+Revises: ce_0035
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+from core.db.task_number_migration import downgrade_task_numbers, upgrade_task_numbers
+
+revision = "ce_0036"
+down_revision = "ce_0035"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ # CE's initial create_all may already contain the current model tables.
+ if "task_number_counters" in sa.inspect(op.get_bind()).get_table_names():
+ return
+ upgrade_task_numbers(op)
+
+
+def downgrade():
+ downgrade_task_numbers(op)
diff --git a/src/backend/api/routes/v1/models.py b/src/backend/api/routes/v1/models.py
index d259055d..17376977 100644
--- a/src/backend/api/routes/v1/models.py
+++ b/src/backend/api/routes/v1/models.py
@@ -532,7 +532,7 @@ def get_main_capabilities(
# context-usage gauge reflects the model's true window instead of a guess.
# 0 when unconfigured — the frontend falls back to a heuristic/default.
main_context_length = int(cfg.context_length) if cfg and cfg.context_length else 0
- from core.files.artifacts.reader import ATTACHMENT_PREVIEW_MAX_CHARS
+ from core.llm.runtime.file_context import ATTACHMENT_HANDLE_EST_CHARS
# Image reading: either the main model is natively multimodal, or a `vision` role
# model is assigned and the vision bridge transcribes images into text evidence.
# The frontend uses `can_read_image` alone to decide whether to offer image upload;
@@ -547,7 +547,7 @@ def get_main_capabilities(
**effort_caps,
"supports_vision": main_native_vision,
"context_length": main_context_length,
- "attachment_preview_chars": ATTACHMENT_PREVIEW_MAX_CHARS,
+ "attachment_preview_chars": ATTACHMENT_HANDLE_EST_CHARS,
},
"vision": {
"can_read_image": vision_cfg is not None,
diff --git a/src/backend/core/chat/compaction/history.py b/src/backend/core/chat/compaction/history.py
index 6ab148eb..d1630548 100644
--- a/src/backend/core/chat/compaction/history.py
+++ b/src/backend/core/chat/compaction/history.py
@@ -8,7 +8,11 @@
from core.chat.context import build_effective_user_message
from core.llm.context import compaction as C
from core.llm.context.ir_constants import SESSION_CONTEXT_META_KEY
-from core.llm.runtime.model_steps import assistant_row_replay, record_assistant_step
+from core.llm.runtime.model_steps import (
+ assistant_row_replay,
+ record_assistant_step,
+ split_before_input,
+)
from core.chat.references import render_reference_block
@@ -56,6 +60,9 @@ def normalize_rows(rows: List[Any], live_run_ids: frozenset = frozenset()) -> Li
definition and would show the model a truncated version of its own reply.
"""
out: List[Dict[str, Any]] = []
+ # Where the latest user row starts: blocks the run injected ahead of the
+ # user's message (attachment previews) go back in front of it.
+ turn_start = 0
for msg in rows:
role = getattr(msg, "role", None)
extra = getattr(msg, "extra_data", None) or {}
@@ -79,9 +86,10 @@ def normalize_rows(rows: List[Any], live_run_ids: frozenset = frozenset()) -> Li
}
if isinstance(extra.get(SESSION_CONTEXT_META_KEY), dict):
row[SESSION_CONTEXT_META_KEY] = dict(extra[SESSION_CONTEXT_META_KEY])
+ turn_start = len(out)
out.append(row)
elif role == "assistant":
- out.extend(
+ before, after = split_before_input(
assistant_row_replay(
content=content or "",
model_steps=(
@@ -92,6 +100,8 @@ def normalize_rows(rows: List[Any], live_run_ids: frozenset = frozenset()) -> Li
cancelled=bool(extra.get("cancelled")),
)
)
+ out[turn_start:turn_start] = before
+ out.extend(after)
else:
row = {"role": role or "user", "content": content}
if isinstance(extra.get(SESSION_CONTEXT_META_KEY), dict):
diff --git a/src/backend/core/chat/runtime/chunks.py b/src/backend/core/chat/runtime/chunks.py
index 544b62a5..b575a03d 100644
--- a/src/backend/core/chat/runtime/chunks.py
+++ b/src/backend/core/chat/runtime/chunks.py
@@ -6,13 +6,12 @@
from core.infra.text_bounds import bound_result_for_display
from core.chat.tool_log import (
attach_subagent_step,
+ build_interaction_event,
build_thinking_event,
build_tool_call_delta_event,
build_tool_call_event,
build_tool_call_start_event,
build_tool_result_event,
- build_user_question_event,
- build_user_question_resolved_event,
)
from core.infra.logging import get_logger
@@ -131,34 +130,8 @@ async def consume_chunk(turn, chunk):
logger.debug("attach_subagent_step failed (ignored)", exc_info=True)
chunk["chat_id"] = turn.chat_id
await turn.emit(bound_result_for_display(chunk)[0])
- elif chunk_type == "file_confirm":
- await turn.emit(
- {
- "type": "file_confirm",
- "chat_id": turn.chat_id,
- "confirm_id": chunk.get("confirm_id"),
- "op": chunk.get("op"),
- "logical_path": chunk.get("logical_path"),
- "message": chunk.get("message"),
- "expired": chunk.get("expired", False),
- }
- )
- elif chunk_type == "design_pick":
- await turn.emit(
- {
- "type": "design_pick",
- "chat_id": turn.chat_id,
- "confirm_id": chunk.get("confirm_id"),
- "question": chunk.get("question", ""),
- "options": chunk.get("options", []),
- "message": chunk.get("message"),
- "expired": chunk.get("expired", False),
- }
- )
- elif chunk_type == "user_question":
- await turn.emit(build_user_question_event(chunk, turn.chat_id))
- elif chunk_type == "user_question_resolved":
- await turn.emit(build_user_question_resolved_event(chunk, turn.chat_id))
+ elif chunk_type in ("file_confirm", "design_pick", "user_question", "user_question_resolved"):
+ await turn.emit(build_interaction_event(chunk_type, chunk, turn.chat_id))
elif chunk_type == "plan_update":
await turn.emit(
{
diff --git a/src/backend/core/chat/steer.py b/src/backend/core/chat/steer.py
index 5dc37396..991f3ab6 100644
--- a/src/backend/core/chat/steer.py
+++ b/src/backend/core/chat/steer.py
@@ -15,6 +15,7 @@
from core.db.engine import SessionLocal
from core.db.models import ChatRun
+from core.harness.request_links import link_request
from core.infra.ephemeral import get_ephemeral_state
from core.chat.steer_queue import SteerQueue, SteerQueueConflict
@@ -65,6 +66,7 @@ async def put_pending_steer(run_id: str, payload: Dict[str, Any]) -> Dict[str, A
delivery_mode=str(payload.get("delivery_mode") or "steer"),
replace_latest=bool(payload.get("replace_latest", True)),
)
+ await link_request(run_id, f"steer:{record.delivery_mode}")
await _notify(
run_id,
{
diff --git a/src/backend/core/chat/tool_log.py b/src/backend/core/chat/tool_log.py
index 0f368009..6d1ee3f1 100644
--- a/src/backend/core/chat/tool_log.py
+++ b/src/backend/core/chat/tool_log.py
@@ -164,28 +164,11 @@ def build_tool_call_event(chunk: dict, chat_id: str, tool_calls_log: list) -> Di
return {"type": "tool_call", **tc, "chat_id": chat_id}
-def build_user_question_event(chunk: dict, chat_id: str) -> Dict[str, Any]:
- """Build the authoritative pending-question request sent to the composer."""
+def build_interaction_event(event_type: str, chunk: dict, chat_id: str) -> Dict[str, Any]:
+ """Relay a confirm / design-pick / question signal; its producer owns the fields."""
- return {
- "type": "user_question",
- "chat_id": chat_id,
- "request_id": chunk.get("request_id"),
- "questions": chunk.get("questions", []),
- "created_at": chunk.get("created_at"),
- "expires_at": chunk.get("expires_at"),
- }
-
-
-def build_user_question_resolved_event(chunk: dict, chat_id: str) -> Dict[str, Any]:
- """Build the server-owned event that removes a composer request."""
-
- return {
- "type": "user_question_resolved",
- "chat_id": chat_id,
- "request_id": chunk.get("request_id"),
- "outcome": chunk.get("outcome"),
- }
+ public = {k: v for k, v in chunk.items() if not k.startswith("_") and k != "event"}
+ return bound_result_for_display({**public, "type": event_type, "chat_id": chat_id})[0]
def _payload_carries_error(res: Any) -> bool:
diff --git a/src/backend/core/db/models/__init__.py b/src/backend/core/db/models/__init__.py
index beb8116f..2ba29242 100644
--- a/src/backend/core/db/models/__init__.py
+++ b/src/backend/core/db/models/__init__.py
@@ -9,7 +9,7 @@
from core.db.edition import EDITION_MODEL_EXPORTS
from core.db.models.chat import ChatCompactionState, ChatMessage, ChatSession
-from core.db.models.chat_runs import ChatRun, ChatRunOperation, ChatSteerQueueItem
+from core.db.models.chat_runs import ChatRun, ChatRunOperation, ChatSteerQueueItem, TaskNumberCounter
from core.db.models.chat_resources import ChatSandboxSnapshot, ToolMediaBlob, MessageFeedback
from core.db.models.identity import LocalUser, UserApiKey, UserFolder, UserShadow
from core.db.models.project import Project, ProjectFavorite
@@ -88,6 +88,7 @@ def __getattr__(name):
'ChatMessage',
'ChatCompactionState',
'ChatRun',
+ 'TaskNumberCounter',
'CommandTask',
'ChatRunOperation',
'ChatSteerQueueItem',
diff --git a/src/backend/core/db/models/chat_runs.py b/src/backend/core/db/models/chat_runs.py
index e0afa2f3..c77cf1c8 100644
--- a/src/backend/core/db/models/chat_runs.py
+++ b/src/backend/core/db/models/chat_runs.py
@@ -3,7 +3,9 @@
from core.db.utc_datetime import UTCDateTime
from datetime import datetime, timezone
from core.db.engine import Base
-from sqlalchemy import BigInteger, CheckConstraint, Column, ForeignKey, Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy import BigInteger, CheckConstraint, Column, ForeignKey, Index, Integer, String, Text, UniqueConstraint, event
+from sqlalchemy.dialects.postgresql import insert as postgresql_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.orm import deferred
from core.db.models.column_types import BigIntPK, JSONType
@@ -18,6 +20,11 @@ class ChatRun(Base):
__tablename__ = "chat_runs"
run_id = Column(String(64), primary_key=True)
+ # The task number people see and quote: RW-YYYYMMDD-NNNN, assigned on insert (see below).
+ # run_id stays the internal key every table joins on.
+ task_no = Column(String(32), nullable=False)
+ # Last sub-task number handed out under this task (RW-…-NNNN-01, -02, …).
+ subtask_seq = Column(Integer, nullable=False, default=0, server_default="0")
chat_id = Column(
String(64),
ForeignKey("chat_sessions.chat_id", ondelete="CASCADE"),
@@ -77,6 +84,7 @@ class ChatRun(Base):
name="chat_runs_writer_slot_check",
),
UniqueConstraint("chat_id", "writer_slot", name="uq_chat_runs_writer_slot"),
+ UniqueConstraint("task_no", name="uq_chat_runs_task_no"),
Index("idx_chat_runs_chat_status", "chat_id", "status"),
Index("idx_chat_runs_status_lease", "status", "lease_expires_at"),
Index("idx_chat_runs_user_id", "user_id"),
@@ -84,6 +92,54 @@ class ChatRun(Base):
)
+class TaskNumberCounter(Base):
+ """The last task serial number handed out on each day (server local time)."""
+
+ __tablename__ = "task_number_counters"
+
+ day = Column(String(8), primary_key=True) # YYYYMMDD
+ last_seq = Column(Integer, nullable=False)
+
+
+TASK_NUMBER_PREFIX = "RW"
+
+
+def task_number(day: str, seq: int) -> str:
+ return f"{TASK_NUMBER_PREFIX}-{day}-{seq:04d}"
+
+
+def subtask_number(parent_no: str, seq: int) -> str:
+ return f"{parent_no}-{seq:02d}"
+
+
+def reserve_task_number(connection, at: datetime) -> str:
+ """Hand out the next task number of ``at``'s day in one atomic upsert.
+
+ The counter row stays locked until the caller's transaction ends, so numbers are
+ gap-free and never collide, on PostgreSQL and SQLite alike.
+ """
+ day = f"{at.astimezone():%Y%m%d}"
+ table = TaskNumberCounter.__table__
+ insert = postgresql_insert if connection.dialect.name == "postgresql" else sqlite_insert
+ statement = (
+ insert(table)
+ .values(day=day, last_seq=1)
+ .on_conflict_do_update(index_elements=[table.c.day], set_={"last_seq": table.c.last_seq + 1})
+ .returning(table.c.last_seq)
+ )
+ return task_number(day, connection.execute(statement).scalar_one())
+
+
+@event.listens_for(ChatRun, "before_insert")
+def assign_task_number(_mapper, connection, target: ChatRun) -> None:
+ """Every run gets its task number where the row is written, whichever path admits it."""
+ if target.task_no is None:
+ at = target.created_at or datetime.now(timezone.utc)
+ if at.tzinfo is None:
+ at = at.replace(tzinfo=timezone.utc)
+ target.task_no = reserve_task_number(connection, at)
+
+
class ChatRunOperation(Base):
"""Append-only, per-run operation journal used for recovery decisions."""
diff --git a/src/backend/core/db/models/logs.py b/src/backend/core/db/models/logs.py
index f88dcf87..0d486b42 100644
--- a/src/backend/core/db/models/logs.py
+++ b/src/backend/core/db/models/logs.py
@@ -287,6 +287,11 @@ class SubAgentCallLog(Base):
error_message = Column(Text)
duration_ms = Column(Integer)
parent_subagent_log_id = Column(String(64), index=True)
+ # Sub-task number: the parent's number (the run's task number, or the calling sub-agent's
+ # own sub-task number) plus "-NN". Agents that are the run itself carry the run's number.
+ task_no = Column(String(64), index=True)
+ # Last sub-task number handed out to sub-agents this one called.
+ child_seq = Column(Integer, nullable=False, default=0, server_default="0")
started_at = Column(UTCDateTime(timezone=True))
completed_at = Column(UTCDateTime(timezone=True))
created_at = Column(UTCDateTime(timezone=True), default=utc_now, nullable=False)
diff --git a/src/backend/core/db/task_number_migration.py b/src/backend/core/db/task_number_migration.py
new file mode 100644
index 00000000..33d1d779
--- /dev/null
+++ b/src/backend/core/db/task_number_migration.py
@@ -0,0 +1,125 @@
+"""Schema and backfill for task numbers, shared by the EE (tasknumber01) and CE (ce_0036) chains.
+
+Adds ``chat_runs.task_no`` / ``subtask_seq``, ``subagent_call_logs.task_no`` / ``child_seq`` and
+the per-day ``task_number_counters`` table, then numbers existing rows by the rules new rows
+follow (``core.db.models.chat_runs`` and ``core.harness.log_records``): runs in creation order by
+local day; sub-agent calls under whoever dispatched them; the run's own agent keeps its number.
+"""
+
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from datetime import timezone
+
+import sqlalchemy as sa
+
+from core.db.utc_datetime import UTCDateTime
+
+_RUN_AGENT_TYPES = {"plan_mode", "user_agent_direct"}
+_BATCH = 1000
+
+_runs = sa.table(
+ "chat_runs",
+ sa.column("run_id", sa.String),
+ sa.column("message_id", sa.String),
+ sa.column("created_at", UTCDateTime(timezone=True)),
+ sa.column("task_no", sa.String),
+ sa.column("subtask_seq", sa.Integer),
+)
+_logs = sa.table(
+ "subagent_call_logs",
+ sa.column("id", sa.String),
+ sa.column("message_id", sa.String),
+ sa.column("subagent_type", sa.String),
+ sa.column("parent_subagent_log_id", sa.String),
+ sa.column("created_at", UTCDateTime(timezone=True)),
+ sa.column("task_no", sa.String),
+ sa.column("child_seq", sa.Integer),
+)
+_counters = sa.table("task_number_counters", sa.column("day", sa.String), sa.column("last_seq", sa.Integer))
+
+
+def _local_day(at) -> str:
+ if at.tzinfo is None:
+ at = at.replace(tzinfo=timezone.utc)
+ return f"{at.astimezone():%Y%m%d}"
+
+
+def _update(bind, table, key, values_by_key, **columns) -> None:
+ statement = table.update().where(table.c[key] == sa.bindparam("_key")).values(
+ **{name: sa.bindparam(f"_{name}") for name in columns}
+ )
+ batch = [
+ {"_key": row_key, **{f"_{name}": values[row_key] for name, values in columns.items()}}
+ for row_key in values_by_key
+ ]
+ for start in range(0, len(batch), _BATCH):
+ bind.execute(statement, batch[start:start + _BATCH])
+
+
+def _backfill(bind) -> None:
+ run_no, run_by_message, per_day = {}, {}, Counter()
+ for run_id, message_id, created_at in bind.execute(
+ sa.select(_runs.c.run_id, _runs.c.message_id, _runs.c.created_at)
+ .order_by(_runs.c.created_at, _runs.c.run_id)
+ ):
+ day = _local_day(created_at)
+ per_day[day] += 1
+ run_no[run_id] = f"RW-{day}-{per_day[day]:04d}"
+ run_by_message[message_id] = run_id
+ if per_day:
+ bind.execute(_counters.insert(), [{"day": d, "last_seq": n} for d, n in per_day.items()])
+
+ log_no, log_seq, run_seq = {}, defaultdict(int), defaultdict(int)
+ for log_id, message_id, kind, parent_id in bind.execute(
+ sa.select(_logs.c.id, _logs.c.message_id, _logs.c.subagent_type, _logs.c.parent_subagent_log_id)
+ .where(_logs.c.message_id.isnot(None))
+ .order_by(_logs.c.created_at, _logs.c.id)
+ ):
+ run_id = run_by_message.get(message_id)
+ if run_id is None:
+ continue
+ if kind in _RUN_AGENT_TYPES:
+ log_no[log_id] = run_no[run_id]
+ elif parent_id in log_no:
+ log_seq[parent_id] += 1
+ log_no[log_id] = f"{log_no[parent_id]}-{log_seq[parent_id]:02d}"
+ else:
+ run_seq[run_id] += 1
+ log_no[log_id] = f"{run_no[run_id]}-{run_seq[run_id]:02d}"
+
+ _update(bind, _runs, "run_id", run_no, task_no=run_no, subtask_seq=defaultdict(int, run_seq))
+ _update(bind, _logs, "id", log_no, task_no=log_no, child_seq=defaultdict(int, log_seq))
+
+
+def upgrade_task_numbers(op) -> None:
+ op.create_table(
+ "task_number_counters",
+ sa.Column("day", sa.String(8), primary_key=True),
+ sa.Column("last_seq", sa.Integer(), nullable=False),
+ )
+ with op.batch_alter_table("chat_runs") as batch:
+ batch.add_column(sa.Column("task_no", sa.String(32), nullable=True))
+ batch.add_column(sa.Column("subtask_seq", sa.Integer(), nullable=False, server_default="0"))
+ with op.batch_alter_table("subagent_call_logs") as batch:
+ batch.add_column(sa.Column("task_no", sa.String(64), nullable=True))
+ batch.add_column(sa.Column("child_seq", sa.Integer(), nullable=False, server_default="0"))
+
+ _backfill(op.get_bind())
+
+ with op.batch_alter_table("chat_runs") as batch:
+ batch.alter_column("task_no", existing_type=sa.String(32), nullable=False)
+ batch.create_unique_constraint("uq_chat_runs_task_no", ["task_no"])
+ op.create_index("ix_subagent_call_logs_task_no", "subagent_call_logs", ["task_no"])
+
+
+def downgrade_task_numbers(op) -> None:
+ op.drop_index("ix_subagent_call_logs_task_no", table_name="subagent_call_logs")
+ with op.batch_alter_table("subagent_call_logs") as batch:
+ batch.drop_column("child_seq")
+ batch.drop_column("task_no")
+ with op.batch_alter_table("chat_runs") as batch:
+ batch.drop_constraint("uq_chat_runs_task_no", type_="unique")
+ batch.drop_column("subtask_seq")
+ batch.drop_column("task_no")
+ op.drop_table("task_number_counters")
diff --git a/src/backend/core/files/artifacts/reader.py b/src/backend/core/files/artifacts/reader.py
index 7c04def2..5107d38e 100644
--- a/src/backend/core/files/artifacts/reader.py
+++ b/src/backend/core/files/artifacts/reader.py
@@ -13,12 +13,6 @@
logger = logging.getLogger(__name__)
-# Maximum parsed-text preview injected automatically for a newly attached file.
-# Full content remains available through read_artifact and is never placed in
-# the request body by the frontend.
-ATTACHMENT_PREVIEW_MAX_CHARS = 5_000
-
-
# Canonical source values stored in Artifact.extra_data["source"].
SOURCE_USER_UPLOAD = "user_upload"
SOURCE_AI_GENERATED = "ai_generated"
diff --git a/src/backend/core/files/content/content_blocks_constants.py b/src/backend/core/files/content/content_blocks_constants.py
index c46a3a41..60288760 100644
--- a/src/backend/core/files/content/content_blocks_constants.py
+++ b/src/backend/core/files/content/content_blocks_constants.py
@@ -103,6 +103,8 @@
"content_label": "内容管理",
"config_label": "系统配置",
"apidoc_label": "接口文档",
+ # Browser tab icon of the admin pages; empty = follow branding.favicon_url.
+ "favicon_url": "",
},
# 与前端 utils/pageConfigDefaults.ts 的 DEFAULT_SIDEBAR_ITEMS / DEFAULT_MENU_ITEMS 保持一致。
# 这只是新装环境的出厂值——已存在的部署以 DB 现有取值为准,由 backfill_navigation_entries
@@ -124,13 +126,6 @@
"dialog_logout_confirm_ok": "退出登录",
"recommend_banner_text": "",
},
- "defaults": {
- # Initial mode on every user login / new chat:
- # - chat_mode: 'turbo' / 'fast' / 'medium' / 'high' / 'max' (recommended)
- # - thinking_mode: bool (legacy field, fallback compat; true→medium, false→fast)
- "chat_mode": "fast",
- "thinking_mode": False,
- },
"auth": {
# Whether the login page shows the "register" entry (tab + register subpage).
# When off, the login page keeps only login, and backend register submissions
diff --git a/src/backend/core/harness/log_records.py b/src/backend/core/harness/log_records.py
index 66dc0eda..2fe60128 100644
--- a/src/backend/core/harness/log_records.py
+++ b/src/backend/core/harness/log_records.py
@@ -1,9 +1,11 @@
"""Synchronous writers for tool, sub-agent and skill call logs."""
from __future__ import annotations
-from typing import Any, Dict
+from typing import Any, Dict, Optional
+from sqlalchemy import select, update
from sqlalchemy.orm import Session
from core.db.engine import SessionLocal
-from core.db.models import SkillCallLog, SubAgentCallLog, ToolCallLog, ToolEffectLedger
+from core.db.models import ChatRun, SkillCallLog, SubAgentCallLog, ToolCallLog, ToolEffectLedger
+from core.db.models.chat_runs import subtask_number
from core.infra.data_masking import mask_sensitive_data
from core.infra.logging import get_logger
from core.harness import log_context
@@ -87,6 +89,42 @@ def write_tool_call_sync(record: Dict[str, Any]) -> None:
db.close()
+# Agents that are the run itself (a plan's coordinator, a dedicated agent chat), not a sub-task.
+RUN_AGENT_TYPES = frozenset({"plan_mode", "user_agent_direct"})
+
+
+def _reserve_subtask_no(db: Session, record: Dict[str, Any]) -> Optional[str]:
+ """Number a sub-agent call under whoever dispatched it, in one atomic UPDATE.
+
+ A sub-agent called by another sub-agent is numbered under that sub-agent; otherwise it is
+ numbered under the run. Agents that are the run itself carry the run's own number.
+ """
+ run_id = record.get("run_id")
+ if record.get("subagent_type") in RUN_AGENT_TYPES:
+ if not run_id:
+ return None
+ return db.execute(select(ChatRun.task_no).where(ChatRun.run_id == run_id)).scalar_one_or_none()
+ row = None
+ parent_id = record.get("parent_subagent_log_id")
+ if parent_id:
+ logs = SubAgentCallLog.__table__
+ row = db.execute(
+ update(logs)
+ .where(logs.c.id == parent_id, logs.c.task_no.is_not(None))
+ .values(child_seq=logs.c.child_seq + 1)
+ .returning(logs.c.task_no, logs.c.child_seq)
+ ).first()
+ if row is None and run_id:
+ runs = ChatRun.__table__
+ row = db.execute(
+ update(runs)
+ .where(runs.c.run_id == run_id)
+ .values(subtask_seq=runs.c.subtask_seq + 1)
+ .returning(runs.c.task_no, runs.c.subtask_seq)
+ ).first()
+ return subtask_number(row[0], row[1]) if row else None
+
+
def write_subagent_create_sync(record: Dict[str, Any]) -> None:
if not log_writer_constants.SUBAGENT_LOG_ENABLED:
return
@@ -112,6 +150,7 @@ def write_subagent_create_sync(record: Dict[str, Any]) -> None:
input_messages=input_safe,
status="running",
parent_subagent_log_id=record.get("parent_subagent_log_id"),
+ task_no=_reserve_subtask_no(db, record),
started_at=record.get("started_at") or log_context.now_utc(),
created_at=log_context.now_utc(),
)
diff --git a/src/backend/core/harness/log_writer.py b/src/backend/core/harness/log_writer.py
index a4ccb852..08effa2d 100644
--- a/src/backend/core/harness/log_writer.py
+++ b/src/backend/core/harness/log_writer.py
@@ -3,7 +3,7 @@
import asyncio
from typing import Any, Dict, List, Optional
from core.infra.background import spawn
-from core.infra.logging import get_logger
+from core.infra.logging import get_logger, run_id_var
from core.harness import log_audit
from core.harness import log_context
from core.harness import log_records
@@ -51,6 +51,7 @@ async def start_subagent_log(record: Dict[str, Any]) -> str:
record.setdefault("chat_id", ctx["chat_id"])
record.setdefault("message_id", log_context.current_message_id.get())
record.setdefault("parent_subagent_log_id", log_context.current_subagent_log_id())
+ record.setdefault("run_id", run_id_var.get()) # numbers the call under its run's task number
record.setdefault("started_at", log_context.now_utc())
try:
await asyncio.to_thread(log_records.write_subagent_create_sync, record)
diff --git a/src/backend/core/harness/request_links.py b/src/backend/core/harness/request_links.py
new file mode 100644
index 00000000..43529b91
--- /dev/null
+++ b/src/backend/core/harness/request_links.py
@@ -0,0 +1,61 @@
+"""Which HTTP requests acted on a run: the bridge from a run_id to request trace_ids.
+
+A run outlives the request that started it and is later steered, answered or cancelled by
+other requests, each with its own trace_id. Every such request appends one ``request`` event
+to the run's append-only event log, so a run's execution trace can list the trace_ids to look
+up in the application log. Code running inside a run's own worker task is not a request and
+never records a link, even though the task inherited the starting request's context.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from core.db.models import HarnessEventLog
+from core.db.threads import run_db
+from core.harness.events import Event
+from core.harness.ledger import DurableEventStore
+from core.infra.logging import get_logger, run_id_var, trace_id_var
+
+logger = get_logger(__name__)
+
+REQUEST_EVENT = "request"
+
+
+def current_request_trace() -> str:
+ """The trace_id of the HTTP request being served; empty inside a run's worker task."""
+ return "" if run_id_var.get() else trace_id_var.get()
+
+
+async def link_request(run_id: str, action: str, *, trace_id: Optional[str] = None) -> None:
+ """Record that the current request (or ``trace_id``) performed ``action`` on ``run_id``.
+
+ Best effort: a missing link only costs the operator a log lookup, so it never fails the
+ request that acted on the run.
+ """
+ trace_id = current_request_trace() if trace_id is None else trace_id
+ if not run_id or not trace_id:
+ return
+ event = Event.create(
+ run_id=run_id, event_type=REQUEST_EVENT, phase="request",
+ payload={"trace_id": trace_id, "action": action},
+ )
+ try:
+ await run_db(DurableEventStore(), event)
+ except Exception as exc: # noqa: BLE001
+ logger.warning("run_request_link_failed", run_id=run_id, action=action, error=str(exc))
+
+
+def list_request_links(db: Session, run_id: str) -> List[Dict[str, Any]]:
+ rows = (
+ db.query(HarnessEventLog.payload, HarnessEventLog.created_at)
+ .filter(HarnessEventLog.run_id == run_id, HarnessEventLog.event_type == REQUEST_EVENT)
+ .order_by(HarnessEventLog.event_seq)
+ .all()
+ )
+ return [
+ {"trace_id": payload.get("trace_id"), "action": payload.get("action"), "at": created_at}
+ for payload, created_at in rows
+ ]
diff --git a/src/backend/core/infra/logging.py b/src/backend/core/infra/logging.py
index 391e76ee..b6106c9c 100644
--- a/src/backend/core/infra/logging.py
+++ b/src/backend/core/infra/logging.py
@@ -44,6 +44,9 @@ def mask_log_data(data):
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="")
user_id_var: ContextVar[str] = ContextVar("user_id", default="")
chat_id_var: ContextVar[str] = ContextVar("chat_id", default="")
+# Set for the lifetime of a run's worker task (core.runs.runtime.register_run_task), so every log
+# line a run produces can be found by its run_id; trace_id stays the id of one HTTP request.
+run_id_var: ContextVar[str] = ContextVar("run_id", default="")
_QUIET_ACCESS_LOG = "__quiet_access_log__"
@@ -77,14 +80,11 @@ def add_trace_id(logger: Any, method_name: str, event_dict: EventDict) -> EventD
def add_user_context(logger: Any, method_name: str, event_dict: EventDict) -> EventDict:
- """Add user_id and chat_id to log event if available."""
- user_id = user_id_var.get()
- chat_id = chat_id_var.get()
-
- if user_id:
- event_dict["user_id"] = user_id
- if chat_id:
- event_dict["chat_id"] = chat_id
+ """Add user_id, chat_id and run_id to log event if available."""
+ for key, var in (("user_id", user_id_var), ("chat_id", chat_id_var), ("run_id", run_id_var)):
+ value = var.get()
+ if value:
+ event_dict.setdefault(key, value)
return event_dict
diff --git a/src/backend/core/llm/context/attachment_hook.py b/src/backend/core/llm/context/attachment_hook.py
index b9c71d19..09714812 100644
--- a/src/backend/core/llm/context/attachment_hook.py
+++ b/src/backend/core/llm/context/attachment_hook.py
@@ -59,7 +59,7 @@ async def __call__(self, invocation: Invocation) -> Decision:
text = file_context.build_historical_files_context(historical)
if text:
items.append(payload(text, kind="project_material", origin="workspace:historical_files", trust="workspace"))
- text = await asyncio.to_thread(file_context.build_file_context, uploaded, user_id=user_id) if uploaded else ""
+ text = file_context.build_file_context(uploaded) if uploaded else ""
if text:
items.append(payload(text, kind="attachment", origin="user:uploaded_files", trust="user"))
images = [file for file in uploaded if artifact_bytes.is_image(file)]
diff --git a/src/backend/core/llm/runtime/agentscope_hook_adapter.py b/src/backend/core/llm/runtime/agentscope_hook_adapter.py
index 25576bac..32820cb8 100644
--- a/src/backend/core/llm/runtime/agentscope_hook_adapter.py
+++ b/src/backend/core/llm/runtime/agentscope_hook_adapter.py
@@ -11,7 +11,7 @@
from core.harness.hooks import HookBus, HookStage, Invocation
from core.llm.runtime.model_steps import record_assistant_step
from core.models.reasoning_replay import stamp_reasoning_origin
-from core.llm.runtime.context_hook_adapter import attachment_input, begin_context_turn, append_hook_context, apply_reasoning_hooks
+from core.llm.runtime.context_hook_adapter import attachment_input, begin_context_turn, append_hook_context, apply_reasoning_hooks, record_hook_context
from core.llm.runtime.model_selection_adapter import apply_model_selection, model_selection_input
from core.harness.ledger import DurableEventStore, HarnessUsageLedger
@@ -241,7 +241,9 @@ async def on_reply(self, agent: Any, input_kwargs: dict, next_handler):
state.context = _restore_context(
getattr(state, "context", None), transformed.data["context"]
)
- append_hook_context(agent, transformed)
+ await record_hook_context(
+ agent, append_hook_context(agent, transformed), before_input=True
+ )
from core.llm.runtime.context_hook_adapter import finish_context_hooks
while True:
final_output = None
diff --git a/src/backend/core/llm/runtime/context_hook_adapter.py b/src/backend/core/llm/runtime/context_hook_adapter.py
index 9989a268..4f782cfb 100644
--- a/src/backend/core/llm/runtime/context_hook_adapter.py
+++ b/src/backend/core/llm/runtime/context_hook_adapter.py
@@ -20,17 +20,49 @@ def begin_context_turn():
def append_hook_context(agent, invocation):
+ appended = []
for item in invocation.data.get("context_items", ()):
item = thaw_value(item)
sequence = next_context_sequence(agent.state.context)
origin = item["origin"]
- append_state_context_item(agent.state, ContextItem.create(
+ context_item = ContextItem.create(
item_id=f"{origin}:{sequence}", kind=item["kind"], origin=origin,
trust=item["trust"], visibility="model", priority=item["priority"],
token_budget=item["token_budget"], truncation_policy=item.get("truncation_policy", POLICY_HEAD_TAIL),
content=item["content"], cache_class="dynamic", created_seq=sequence,
render_role="user", message_group=f"{origin}:{sequence}",
- ))
+ )
+ append_state_context_item(agent.state, context_item)
+ appended.append(context_item)
+ return appended
+
+
+# A steer is persisted as its own user row by the chat runtime; recording the
+# injection as well would replay the instruction twice.
+_NOT_RECORDED_KINDS = frozenset({"steer"})
+
+
+async def record_hook_context(agent, items, *, before_input):
+ """Keep injected blocks in the turn's step record so the next turn replays them.
+
+ Only the reply and reasoning stages call this: their blocks sit at a clear
+ position (before the user's message / before a model call). Tool-stage
+ injections can land between a call and its result and are not replayed.
+ """
+ import asyncio
+ from core.llm.runtime.model_steps import record_context_step
+
+ ledger = getattr(getattr(agent, "state", None), "pending_model_steps", None)
+ if ledger is None:
+ return
+ for item in items:
+ if item.kind in _NOT_RECORDED_KINDS:
+ continue
+ if isinstance(item.content, (list, tuple)): # may carry media → blob table write
+ step = await asyncio.to_thread(record_context_step, item, before_input=before_input)
+ else:
+ step = record_context_step(item, before_input=before_input)
+ ledger.append(step)
async def apply_reasoning_hooks(adapter, agent, input_kwargs, plain, restore_context):
@@ -49,7 +81,7 @@ async def apply_reasoning_hooks(adapter, agent, input_kwargs, plain, restore_con
)
if transformed.data.get("context") is not None:
state.context = restore_context(state.context, transformed.data["context"])
- append_hook_context(agent, transformed)
+ await record_hook_context(agent, append_hook_context(agent, transformed), before_input=False)
await adapter._stage(HookStage.AFTER_CONTEXT, agent, operation_name="agent_reasoning",
data={"steer_delivery": transformed.data.get("steer_delivery")})
if transformed.data.get("steer_delivery") is not None:
diff --git a/src/backend/core/llm/runtime/file_context.py b/src/backend/core/llm/runtime/file_context.py
index 8f0c26fe..fd54c819 100644
--- a/src/backend/core/llm/runtime/file_context.py
+++ b/src/backend/core/llm/runtime/file_context.py
@@ -1,4 +1,4 @@
-"""Turn file attachments into context: text, images and spreadsheet previews."""
+"""Turn file attachments into context: file handles for documents, base64 for images."""
from __future__ import annotations
from core.files import artifact_bytes
import base64
@@ -26,181 +26,42 @@ def fetch_image_base64(
return base64.b64encode(raw).decode("utf-8"), mime_type
-def is_xlsx(f: Dict[str, Any]) -> bool:
- """Detect xlsx by mime_type or filename extension.
+# Rough size of one attachment handle block, reported to the frontend so its
+# context gauge estimates what the backend actually injects for a staged file.
+ATTACHMENT_HANDLE_EST_CHARS = 300
- We only special-case xlsx (not xls/csv) because:
- - xlsx is the dominant batch-execution upload format
- - parse_xlsx_preview uses openpyxl, same engine as the batch path
- - xls would need a separate xlrd-based preview helper (out of scope)
- """
- mime = (f.get("mime_type") or "").lower()
- if "spreadsheetml.sheet" in mime:
- return True
- name = (f.get("name") or f.get("filename") or "").lower()
- return name.endswith(".xlsx")
+def build_file_context(uploaded_files: List[Dict[str, Any]]) -> str:
+ """Describe the turn's non-image attachments as handles: name, MIME and ``file_id``.
-def build_xlsx_preview_block(
- f: Dict[str, Any],
- user_id: Optional[str] = None,
-) -> Optional[str]:
- """Build the bounded preview + metadata + directive block for a single xlsx.
-
- Returns None if the file can't be read (caller falls back to the regular
- parsed_text path, so the agent at least sees *something*).
- """
- from core.files.content.spreadsheet_parser import parse_xlsx_preview
-
- file_id = f.get("file_id") or ""
- name = f.get("name") or f.get("filename") or "未命名表格"
- if not file_id:
- return None
-
- file_bytes = artifact_bytes.download_artifact_bytes(
- file_id,
- name,
- "xlsx preview",
- user_id=user_id,
- )
- if file_bytes is None:
- return None
-
- try:
- from core.files.artifacts.reader import ATTACHMENT_PREVIEW_MAX_CHARS, fetch_parsed_text
-
- # Reuse the bytes already downloaded for the structural preview to
- # populate parsed_text + summary without a second storage round-trip.
- fetch_parsed_text(file_id, user_id=user_id, prefetched_bytes=file_bytes)
- info = parse_xlsx_preview(file_bytes, char_budget=ATTACHMENT_PREVIEW_MAX_CHARS)
- except RuntimeError as e:
- logger.warning(f"xlsx preview: parse failed for {file_id}: {e}")
- return None
- except Exception as e:
- logger.warning(f"xlsx preview: unexpected parse error for {file_id}: {e}")
- return None
-
- total_rows = info["total_rows"]
- total_cols = info["total_columns"]
- preview_rows = info["preview_rows"]
- omitted = max(0, total_rows - preview_rows)
- cols_str = ", ".join(info["header"]) if info["header"] else "(无列名)"
-
- lines: List[str] = [
- f"[文件: {name}]",
- f" file_id: {file_id}",
- f" 总规模: {total_rows} 行 × {total_cols} 列",
- f" 列名: [{cols_str}]",
- f" 已展示: 前 {preview_rows} 行(约 {len(info['preview_md'])} 字符,达到预览预算)",
- ]
- if info.get("truncated_columns"):
- lines.append(f" 注意: 共 {total_cols} 列,预览仅含前 {len(info['header'])} 列")
- if info.get("truncated_cells"):
- lines.append(f" 注意: {info['truncated_cells']} 个超长单元格已截断到 500 字以内")
- lines.append("")
- lines.append(info["preview_md"])
- if omitted:
- lines.append(f"... (剩余 {omitted} 行未展示) ...")
- lines.append("")
- lines.append("[操作指引]")
- lines.append(
- f"- 需要预览之外的正文时,调用 read_artifact(file_id='{file_id}', "
- "offset=N, limit=4000) 按需读取"
- )
- lines.append("- 以上只是有界预览,不要把未展示的行当作不存在,也不要推断未读取内容")
- return "\n".join(lines)
-
-
-def build_file_context(
- uploaded_files: List[Dict[str, Any]],
- user_id: Optional[str] = None,
-) -> str:
- """Assemble the text-attachment list into the context text injected into the model (non-image files only).
-
- Files are resolved exclusively by ``file_id``. The backend lazily parses and
- caches the artifact, then injects only metadata plus a bounded preview. The
- request body never supplies parsed content.
-
- ``user_id`` is used as the ownership check when fetching ``fetch_parsed_text`` /
- the xlsx preview, preventing a forged request-body ``attachments[].file_id``
- from reading across users — see the :func:`artifact_bytes.download_artifact_bytes` docstring
- for the core defense point.
+ No file is downloaded or parsed here, so attaching a large PDF adds nothing
+ to time-to-first-token. The model reads content on demand through
+ ``read_artifact``, which owns parsing, the ``parsed_text`` cache and the
+ ``user_id`` ownership check.
"""
text_files = [f for f in uploaded_files if not artifact_bytes.is_image(f)]
if not text_files:
return ""
- file_descriptions = []
- for f in text_files:
- name = f.get("name", "未知文件")
- fid = f.get("file_id", "")
- desc = f"- {name}"
- if fid:
- desc += f" (file_id: {fid})"
- file_descriptions.append(desc)
- from core.files.artifacts.reader import ATTACHMENT_PREVIEW_MAX_CHARS, fetch_parsed_text
-
- file_content_parts: List[str] = []
+ lines: List[str] = []
for f in text_files:
- # ── xlsx: bounded preview + metadata + directive (no full content) ──
- if is_xlsx(f):
- block = build_xlsx_preview_block(f, user_id=user_id)
- if block is not None:
- file_content_parts.append(block)
- continue
- # Fall through to regular path on preview failure so the agent
- # at least sees the parsed markdown (degraded but non-fatal).
-
+ name = f.get("name") or f.get("filename") or "未知文件"
fid = (f.get("file_id") or "").strip()
- mime_type = f.get("mime_type") or ""
+ mime_type = f.get("mime_type") or "(未知)"
if not fid:
- file_content_parts.append(
- f"[文件: {f.get('name', '未知文件')}]\n"
- " 状态: 缺少 file_id,无法读取正文"
- )
- continue
-
- logger.info("Document hook: loading preview for %s (%s)", f.get("name"), fid)
- content = fetch_parsed_text(fid, user_id=user_id)
- if not content:
- file_content_parts.append(
- f"[文件: {f.get('name', '未知文件')}]\n"
- f" file_id: {fid}\n"
- f" MIME: {mime_type or '(未知)'}\n"
- " 状态: 文件不可访问或解析失败"
- )
+ lines.append(f"- {name} (MIME: {mime_type}) [缺少 file_id,无法读取]")
continue
-
- total_chars = len(content)
- preview = content[:ATTACHMENT_PREVIEW_MAX_CHARS]
- truncated = total_chars > len(preview)
- block = [
- f"[文件: {f.get('name', '未知文件')}]",
- f" file_id: {fid}",
- f" MIME: {mime_type or '(未知)'}",
- f" 解析文本: 共 {total_chars} 字符,本轮自动展示 {len(preview)} 字符",
- "",
- preview,
- ]
- if truncated:
- block.append(f"... (其余 {total_chars - len(preview)} 字符未展示) ...")
- block.extend(
- [
- "",
- "[读取指引]",
- f"- 需要未展示的正文时,调用 read_artifact(file_id='{fid}', offset=N, limit=4000)",
- "- 不要推断尚未读取的内容",
- ]
- )
- file_content_parts.append("\n".join(block))
- file_content = "\n\n---\n\n".join(file_content_parts)
+ lines.append(f"- {name} (file_id: {fid}, MIME: {mime_type})")
return (
- f"[file name]: {chr(10).join(file_descriptions)}\n"
- f"[file content begin]\n"
- f"{file_content}\n"
- f"[file content end]\n"
- "以上是用户附件的元数据和有界预览;需要预览之外的正文时必须按需调用 read_artifact。"
+ "[用户上传的附件](本轮只提供文件元数据,正文尚未读取)\n"
+ + "\n".join(lines)
+ + "\n\n[读取指引]\n"
+ "- 需要文件内容时调用 read_artifact(file_id, offset=0, limit=4000) 读取解析后的文本,"
+ "按返回的 next_offset 继续分页\n"
+ "- xlsx 先看返回的 sheet_names 再用 sheet_name 读单表;pptx 用 slide_index 读单页\n"
+ "- 调用 read_artifact 之前不要声称已读过附件,也不要推断其内容;"
+ "与附件无关的问题可以不读"
)
diff --git a/src/backend/core/llm/runtime/interaction_store.py b/src/backend/core/llm/runtime/interaction_store.py
index de3911c9..4264fb56 100644
--- a/src/backend/core/llm/runtime/interaction_store.py
+++ b/src/backend/core/llm/runtime/interaction_store.py
@@ -14,7 +14,9 @@
import time
from urllib.parse import quote
+from core.harness.request_links import link_request
from core.infra.ephemeral import get_ephemeral_state
+from core.infra.logging import run_id_var
_PREFIX = "human-interaction:v1:"
_POLL = 0.2
@@ -37,7 +39,8 @@ def _live(record):
async def register(kind, chat_id, request_id, info, timeout):
now = time.time()
record = dict(
- info=info, created_at=now, expires_at=now + timeout, lease_until=now + _LEASE, decision=None
+ info=info, created_at=now, expires_at=now + timeout, lease_until=now + _LEASE, decision=None,
+ run_id=run_id_var.get(), # registered by the waiting tool, inside its run's worker task
)
await get_ephemeral_state().put(
_key(kind, chat_id, request_id), json.dumps(record), ttl=_ttl(record)
@@ -86,6 +89,7 @@ async def decide(kind, chat_id, request_id, decision):
return False
record["decision"] = decision
if await store.compare_exchange(key, raw, json.dumps(record), ttl=_ttl(record)):
+ await link_request(record.get("run_id") or "", f"answer:{kind}")
return True
diff --git a/src/backend/core/llm/runtime/model_steps.py b/src/backend/core/llm/runtime/model_steps.py
index f220198e..e9a2a3a8 100644
--- a/src/backend/core/llm/runtime/model_steps.py
+++ b/src/backend/core/llm/runtime/model_steps.py
@@ -4,7 +4,8 @@
columns (``content`` / ``thinking`` / ``tool_calls`` / ``metadata.segments``)
serve the chat UI. ``model_steps`` is the record the model's own history is
rebuilt from: one entry per model response with the exact blocks it returned,
-and one entry per tool result as it entered the context. Replaying it yields
+one entry per tool result as it entered the context, and one entry per block a
+hook injected into the context (attachment previews, reminders). Replaying it yields
the same message sequence the agent held in memory, so a later turn — or a
resumed one — sends the model what it actually said, step by step, instead of
a reassembly guessed from three separate columns.
@@ -20,7 +21,10 @@
STEP_SCHEMA = "harness.steps.v1"
STEP_ASSISTANT = "assistant"
STEP_TOOL_RESULT = "tool_result"
+STEP_CONTEXT = "context"
STATE_INTERRUPTED = "interrupted"
+# Replay-only marker on a context row that preceded the user's message live.
+BEFORE_INPUT_KEY = "_before_input"
_CANCELLED_TURN_MARKER = "[本轮回答被用户中断]"
_INTERRUPTED_RESULT = "[tool call was interrupted before it returned a result]"
@@ -148,6 +152,61 @@ def record_tool_result_step(block: Any) -> Dict[str, Any]:
}
+def record_context_step(item: Any, *, before_input: bool) -> Dict[str, Any]:
+ """One hook-injected context item, as the model saw it.
+
+ Without this record the next turn rebuilds history without the block, so its
+ request diverges from the cached prefix right there and everything after it
+ is billed again. ``before_input`` marks items added in ``on_reply``, which
+ AgentScope places ahead of the user's message. Media goes to the blob table
+ like tool media, so the caller runs this off the event loop when it has any.
+ """
+ from core.infra.immutable import thaw_json
+ from core.llm.context.items import session_context_metadata
+
+ content = thaw_json(item.content)
+ if isinstance(content, Mapping):
+ content = [content]
+ blocks = (
+ [{"type": "text", "text": str(content or "")}]
+ if not isinstance(content, list)
+ else _persistable_output(content)
+ )
+ return {
+ "schema": STEP_SCHEMA,
+ "kind": STEP_CONTEXT,
+ "before_input": bool(before_input),
+ "context_item": session_context_metadata(item),
+ "blocks": blocks,
+ }
+
+
+def _context_row(step: Mapping[str, Any], blocks: List[Dict[str, Any]]) -> Dict[str, Any]:
+ from core.llm.context.ir_constants import SESSION_CONTEXT_META_KEY
+
+ texts = [b for b in blocks if b.get("type") == "text"]
+ # A text-only item rendered live as one text block; a plain string replays to the same.
+ content: Any = texts[0]["text"] if len(blocks) == 1 and texts else blocks
+ row: Dict[str, Any] = {"role": "user", "content": content}
+ if isinstance(step.get("context_item"), Mapping):
+ row[SESSION_CONTEXT_META_KEY] = dict(step["context_item"])
+ if step.get("before_input"):
+ row[BEFORE_INPUT_KEY] = True
+ return row
+
+
+def split_before_input(
+ rows: Sequence[Mapping[str, Any]],
+) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+ """Separate the rows that preceded the user's message live from the rest."""
+ before: List[Dict[str, Any]] = []
+ after: List[Dict[str, Any]] = []
+ for row in rows:
+ row = dict(row)
+ (before if row.pop(BEFORE_INPUT_KEY, False) else after).append(row)
+ return before, after
+
+
def _ordered_result_steps(
steps: Sequence[Mapping[str, Any]], *, state: str, missing_output: str
) -> List[Dict[str, Any]]:
@@ -287,6 +346,7 @@ def replay_rows(
steps: Sequence[Mapping[str, Any]],
*,
cancelled: bool = False,
+ include_context: bool = True,
) -> List[Dict[str, Any]]:
"""Rebuild the model-facing rows of one assistant message from its record.
@@ -321,6 +381,9 @@ def replay_rows(
rows[-1]["content"].extend(blocks)
else:
rows.append({"role": "tool", "content": blocks})
+ elif kind == STEP_CONTEXT:
+ if include_context:
+ rows.append(_context_row(step, blocks))
else:
raise ValueError(f"unknown model step kind {kind!r}")
if cancelled:
@@ -335,10 +398,11 @@ def assistant_row_replay(
content: str,
model_steps: Any,
cancelled: bool = False,
+ include_context: bool = True,
) -> List[Dict[str, Any]]:
"""Replay canonical steps, or the body of an unstructured text message."""
if isinstance(model_steps, list) and model_steps:
- return replay_rows(model_steps, cancelled=cancelled)
+ return replay_rows(model_steps, cancelled=cancelled, include_context=include_context)
blocks = [{"type": "text", "text": content}] if content else []
if cancelled:
_append_cancel_marker(blocks)
@@ -387,15 +451,19 @@ def explode_history_rows(history: Sequence[Mapping[str, Any]]) -> List[Dict[str,
__all__ = [
+ "BEFORE_INPUT_KEY",
"STATE_INTERRUPTED",
"STEP_ASSISTANT",
+ "STEP_CONTEXT",
"STEP_SCHEMA",
"STEP_TOOL_RESULT",
"assistant_row_replay",
"close_dangling_calls",
"explode_history_rows",
"record_assistant_step",
+ "record_context_step",
"record_tool_result_step",
"replace_final_text",
"replay_rows",
+ "split_before_input",
]
diff --git a/src/backend/core/llm/runtime/tool_media_store.py b/src/backend/core/llm/runtime/tool_media_store.py
index a1f12fc3..3974914b 100644
--- a/src/backend/core/llm/runtime/tool_media_store.py
+++ b/src/backend/core/llm/runtime/tool_media_store.py
@@ -169,9 +169,13 @@ def _media_blocks(rows: Iterable[Any]) -> List[Mapping[str, Any]]:
"""历史行里所有的媒体引用块。"""
found: List[Mapping[str, Any]] = []
for row in rows:
- if not isinstance(row, Mapping):
+ if not isinstance(row, Mapping) or not isinstance(row.get("content"), list):
continue
for block in row.get("content") or []:
+ # 钩子注入在用户消息前的图片(原生视觉附件),直接挂在行内容上。
+ if is_stored(block):
+ found.append(block)
+ continue
if not isinstance(block, Mapping) or block.get("type") != "tool_result":
continue
output = block.get("output")
@@ -221,7 +225,9 @@ def resolved(block: Mapping[str, Any]) -> Dict[str, Any]:
continue
new_blocks = []
for block in blocks:
- if isinstance(block, Mapping) and block.get("type") == "tool_result":
+ if is_stored(block):
+ block = resolved(block)
+ elif isinstance(block, Mapping) and block.get("type") == "tool_result":
output = block.get("output")
if isinstance(output, list):
block = {
diff --git a/src/backend/core/llm/tools/confirmation/gate.py b/src/backend/core/llm/tools/confirmation/gate.py
index afeda91e..8256c7f8 100644
--- a/src/backend/core/llm/tools/confirmation/gate.py
+++ b/src/backend/core/llm/tools/confirmation/gate.py
@@ -115,8 +115,6 @@ def _claim_first(pd: state._Pending, decision: str) -> bool:
cid,
p,
timeout=timeout,
- expire_message="等待确认超时,已自动取消。如仍需要请重新发起。",
- cancel_message="本次运行已取消,该确认已关闭。",
in_lock_extract=_claim_first,
)
if decision in (state.DECISION_ALLOW, state.DECISION_ALLOW_SESSION):
@@ -217,8 +215,6 @@ def _reject_concurrent(st: state._ChatConfirm):
cid,
p,
timeout=timeout,
- expire_message="设计方案选择已超时,助手将自行选择方案继续。",
- cancel_message="本次运行已取消,设计方案选择已关闭。",
in_lock_extract=lambda pd, _d: pd.choice,
)
if decision == state.DECISION_CHOICE and choice:
diff --git a/src/backend/core/llm/tools/confirmation/pending.py b/src/backend/core/llm/tools/confirmation/pending.py
index 3069093f..490f137c 100644
--- a/src/backend/core/llm/tools/confirmation/pending.py
+++ b/src/backend/core/llm/tools/confirmation/pending.py
@@ -84,8 +84,6 @@ async def _wait_pending(
p: state._Pending,
*,
timeout: float,
- expire_message: str,
- cancel_message: str,
in_lock_extract=None,
) -> Tuple[str, Any]:
"""Resolve, claim execution and release the last waiter under one state lock.
@@ -145,8 +143,9 @@ async def _wait_pending(
"logical_path": p.logical_path,
"kind": p.kind,
"expired": True,
- "resolved": not cancelled and decision != state._DECISION_TIMEOUT,
- "message": cancel_message if cancelled else expire_message,
+ "outcome": "cancelled" if cancelled else (
+ "timeout" if decision == state._DECISION_TIMEOUT else "decided"
+ ),
}
if expire_q is not None and expire_payload is not None:
try:
diff --git a/src/backend/core/runs/binding.py b/src/backend/core/runs/binding.py
index 53455534..b1095f5c 100644
--- a/src/backend/core/runs/binding.py
+++ b/src/backend/core/runs/binding.py
@@ -10,6 +10,7 @@
from core.db.engine import SessionLocal
from core.db.models import ChatRun, ChatSession
from core.db.threads import run_db
+from core.infra.logging import run_id_var
from core.runs.journal_domain import DurableRunBinding, RunAlreadyExists, RunJournalError, RunLeaseLost
from sqlalchemy.orm import sessionmaker
@@ -139,6 +140,8 @@ async def _heartbeat() -> None:
heartbeat = asyncio.create_task(_heartbeat())
from core.llm.context.safety import request_admission_scope
+ # The bound work runs in this very task: scope it to the run like a registered run task.
+ run_scope = run_id_var.set(run_id)
with request_admission_scope(
context={"run_id": run_id, "journal_owner": owner},
) as admission:
@@ -189,6 +192,7 @@ async def _heartbeat() -> None:
else:
await run_db(journal.complete, run_id, owner=owner, status="completed")
finally:
+ run_id_var.reset(run_scope)
heartbeat.cancel()
with suppress(asyncio.CancelledError):
await heartbeat
diff --git a/src/backend/core/runs/runtime.py b/src/backend/core/runs/runtime.py
index 76790c07..b7beaa44 100644
--- a/src/backend/core/runs/runtime.py
+++ b/src/backend/core/runs/runtime.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
+import contextvars
import datetime as _datetime
import typing as _typing
from typing import Any, Awaitable, Callable, List, Optional, Tuple
@@ -17,7 +18,9 @@
import core.runs.tasks as tasks
from core.db.models import ChatRun
from core.db.threads import run_db
-from core.infra.logging import get_logger
+from core.harness.request_links import current_request_trace, link_request
+from core.infra.background import spawn
+from core.infra.logging import get_logger, run_id_var
logger = get_logger(__name__)
@@ -31,12 +34,20 @@ class ChatRunPermissionDenied(Exception):
def register_run_task(run_id: str, coro: Awaitable[None], *, name: str) -> None:
- """Spawn the worker coroutine, register it in _active_runs, auto-cleanup on done."""
+ """Spawn the worker coroutine, register it in _active_runs, auto-cleanup on done.
+
+ The task runs with ``run_id`` bound for logging. The request that started it is linked
+ off the run's path, so the link never delays the first model call.
+ """
active = tasks.process_state().active_runs
if run_id in active and not active[run_id].done():
coro.close()
return
- task = asyncio.create_task(coro, name=name)
+ context = contextvars.copy_context()
+ context.run(run_id_var.set, run_id)
+ task = asyncio.create_task(coro, name=name, context=context)
+ if current_request_trace():
+ spawn(link_request(run_id, "start"), name=f"run_request_link:{run_id}")
tasks.process_state().active_runs[run_id] = task
def finished(worker):
@@ -166,6 +177,7 @@ async def cancel_run(
)
if run is None:
return False
+ await link_request(run_id, "cancel")
payload = run.request_payload if isinstance(run.request_payload, dict) else {}
cooperative_only = payload.get("kind") in settings._COOPERATIVE_KINDS
diff --git a/src/backend/orchestration/plans/mode.py b/src/backend/orchestration/plans/mode.py
index ffb885c9..1bc7a476 100644
--- a/src/backend/orchestration/plans/mode.py
+++ b/src/backend/orchestration/plans/mode.py
@@ -370,14 +370,11 @@ def _build_plan_generation_messages(
return messages
-def _build_file_context(
- uploaded_files: List[Dict[str, Any]],
- user_id: Optional[str] = None,
-) -> str:
- """Reuse the main-chat file_id-based lazy preview pipeline."""
- from core.llm.runtime.file_context import build_file_context as build_lazy_file_context
+def _build_file_context(uploaded_files: List[Dict[str, Any]]) -> str:
+ """Reuse the main-chat file_id handle pipeline; content is read via read_artifact."""
+ from core.llm.runtime.file_context import build_file_context as build_handle_context
- return build_lazy_file_context(uploaded_files, user_id=user_id)
+ return build_handle_context(uploaded_files)
def _generation_inputs(
@@ -466,12 +463,7 @@ async def astream_generate_plan(
ontology_runtime=ontology_runtime,
)
- # Worker thread: building the context parses attachments, which on a cache miss
- # is a blocking POST to the external file-parser service. Inline it would pin the
- # event loop and silence every open SSE stream on the instance.
- file_context = await asyncio.to_thread(
- _build_file_context, uploaded_files or [], user_id=user_id
- )
+ file_context = _build_file_context(uploaded_files or [])
# Prepare history from chat session. Plan rules, attachment evidence,
# and the actual user task remain separate canonical candidates.
history = await _prepare_history(session_messages or [], model_name)
@@ -944,10 +936,7 @@ def _cancellation_requested() -> bool:
agent.state.context.extend(session_to_msgs(prepared_history))
# Inject file context (content must be a list of blocks).
- # Worker thread for the same reason as the plan-generation path above.
- file_context = await asyncio.to_thread(
- _build_file_context, uploaded_files or [], user_id=user_id
- )
+ file_context = _build_file_context(uploaded_files or [])
if file_context:
append_context_text(
agent,
diff --git a/src/backend/personal_agent/conversation/front.py b/src/backend/personal_agent/conversation/front.py
index e4bafc4b..22a88f00 100644
--- a/src/backend/personal_agent/conversation/front.py
+++ b/src/backend/personal_agent/conversation/front.py
@@ -196,8 +196,10 @@ def history(db, ident):
elif row.role == "assistant":
if extra.get("model_history_display_only"):
continue
+ # This window truncates and merges rows, so it never matched the live
+ # request byte for byte; injected context rows would only add user turns.
out.extend(assistant_row_replay(
- content=row.content or "", model_steps=row.model_steps,
+ content=row.content or "", model_steps=row.model_steps, include_context=False,
))
continue
else:
diff --git a/src/backend/tests/api/chat_fork_fixtures.py b/src/backend/tests/api/chat_fork_fixtures.py
index 1f759cea..414c5f87 100644
--- a/src/backend/tests/api/chat_fork_fixtures.py
+++ b/src/backend/tests/api/chat_fork_fixtures.py
@@ -3,7 +3,7 @@
import pytest
from core.auth.backend import UserContext, get_current_user
from core.db.engine import Base, get_db
-from core.db.models import ChatCompactionState, ChatMessage, ChatRun, ChatSession, Project
+from core.db.models import ChatCompactionState, ChatMessage, ChatRun, ChatSession, Project, TaskNumberCounter
from core.db.models.command import CommandTask
from fastapi import FastAPI
from fastapi.testclient import TestClient
@@ -27,6 +27,7 @@ def fork_env(tmp_path, monkeypatch):
ChatSession.__table__,
ChatMessage.__table__,
ChatRun.__table__,
+ TaskNumberCounter.__table__,
ChatCompactionState.__table__,
Project.__table__,
CommandTask.__table__,
diff --git a/src/backend/tests/api/test_agent_response_admission.py b/src/backend/tests/api/test_agent_response_admission.py
index 78afac30..450f271a 100644
--- a/src/backend/tests/api/test_agent_response_admission.py
+++ b/src/backend/tests/api/test_agent_response_admission.py
@@ -22,7 +22,7 @@
from core.auth.backend import UserContext
from core.chat.schemas import ChatRequest
from core.db.engine import Base
-from core.db.models import ChatMessage, ChatRun, ChatSession
+from core.db.models import ChatMessage, ChatRun, ChatSession, TaskNumberCounter
from core.chat.sequencer import ChatSequencer
from fastapi import HTTPException
from sqlalchemy import create_engine
@@ -36,7 +36,7 @@ def _route_database(tmp_path, name="route.db"):
)
Base.metadata.create_all(
engine,
- tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__],
+ tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__, TaskNumberCounter.__table__],
)
Session = sessionmaker(bind=engine)
with Session() as db:
@@ -93,7 +93,7 @@ def test_concurrent_requests_have_one_durable_winner(monkeypatch, tmp_path, tran
)
Base.metadata.create_all(
engine,
- tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__],
+ tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__, TaskNumberCounter.__table__],
)
Session = sessionmaker(bind=engine)
with Session() as db:
diff --git a/src/backend/tests/api/test_chat_stream_sequencer.py b/src/backend/tests/api/test_chat_stream_sequencer.py
index 3589d2e8..1b339c8a 100644
--- a/src/backend/tests/api/test_chat_stream_sequencer.py
+++ b/src/backend/tests/api/test_chat_stream_sequencer.py
@@ -19,7 +19,7 @@
import pytest
from core.auth.backend import UserContext
from core.db.engine import Base
-from core.db.models import ChatMessage, ChatRun, ChatSession
+from core.db.models import ChatMessage, ChatRun, ChatSession, TaskNumberCounter
from core.chat.sequencer import ChatSequencer
from fastapi import HTTPException
from sqlalchemy import create_engine
@@ -33,7 +33,7 @@ def _route_database(tmp_path, name="route.db"):
)
Base.metadata.create_all(
engine,
- tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__],
+ tables=[ChatSession.__table__, ChatMessage.__table__, ChatRun.__table__, TaskNumberCounter.__table__],
)
Session = sessionmaker(bind=engine)
with Session() as db:
diff --git a/src/backend/tests/chat/test_stream_event_builders.py b/src/backend/tests/chat/test_stream_event_builders.py
index 2cca620c..83b69b9a 100644
--- a/src/backend/tests/chat/test_stream_event_builders.py
+++ b/src/backend/tests/chat/test_stream_event_builders.py
@@ -13,8 +13,7 @@
build_tool_call_event,
build_tool_call_start_event,
build_tool_result_event,
- build_user_question_event,
- build_user_question_resolved_event,
+ build_interaction_event,
)
@@ -161,8 +160,10 @@ def test_interrupted_tool_result_keeps_neutral_status():
def test_user_question_requested_and_resolved_wire_events():
- requested = build_user_question_event(
+ requested = build_interaction_event(
+ "user_question",
{
+ "event": "requested",
"request_id": "req-1",
"questions": [{"id": "scope", "question": "范围?"}],
"created_at": 10.0,
@@ -179,8 +180,9 @@ def test_user_question_requested_and_resolved_wire_events():
"expires_at": 20.0,
}
- resolved = build_user_question_resolved_event(
- {"request_id": "req-1", "outcome": "answered"},
+ resolved = build_interaction_event(
+ "user_question_resolved",
+ {"event": "resolved", "request_id": "req-1", "outcome": "answered"},
"chat-1",
)
assert resolved == {
diff --git a/src/backend/tests/core/runs/test_request_links.py b/src/backend/tests/core/runs/test_request_links.py
new file mode 100644
index 00000000..6eac028b
--- /dev/null
+++ b/src/backend/tests/core/runs/test_request_links.py
@@ -0,0 +1,118 @@
+"""A run's log scope (run_id) and the HTTP requests (trace_ids) that acted on it."""
+
+from __future__ import annotations
+
+import asyncio
+
+import core.harness.request_links as request_links
+import core.runs.runtime as runtime
+import pytest
+from core.db.engine import Base
+from core.harness.ledger import DurableEventStore
+from core.infra.logging import add_user_context, run_id_var, trace_id_var
+from core.llm.runtime import interaction_store
+from core.runs.journal import RunJournal
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+RUN = "run-links"
+
+
+@pytest.fixture()
+def sessions(tmp_path, monkeypatch):
+ engine = create_engine(
+ f"sqlite:///{tmp_path / 'request-links.sqlite'}", connect_args={"check_same_thread": False}
+ )
+ Base.metadata.create_all(engine)
+ factory = sessionmaker(bind=engine, expire_on_commit=False)
+ RunJournal(factory).accept(
+ run_id=RUN, message_id="msg-links", chat_id="chat-links", user_id="user-links",
+ request_payload={"kind": "chat"}, recovery_snapshot={"kind": "chat", "worker_args": {}},
+ )
+ monkeypatch.setattr(request_links, "DurableEventStore", lambda: DurableEventStore(factory))
+ yield factory
+ engine.dispose()
+
+
+def _links(factory):
+ with factory() as db:
+ return [(link["action"], link["trace_id"]) for link in request_links.list_request_links(db, RUN)]
+
+
+async def _settle_links():
+ pending = [t for t in asyncio.all_tasks() if t.get_name().startswith("run_request_link:")]
+ await asyncio.gather(*pending)
+
+
+@pytest.mark.asyncio
+async def test_run_task_logs_under_its_run_id_and_links_the_starting_request(sessions):
+ seen = {}
+
+ async def worker():
+ seen["run_id"] = run_id_var.get()
+ seen["request"] = request_links.current_request_trace()
+
+ token = trace_id_var.set("trace-start")
+ try:
+ runtime.register_run_task(RUN, worker(), name=f"test:{RUN}")
+ await runtime.tasks.process_state().active_runs[RUN]
+ finally:
+ trace_id_var.reset(token)
+ await _settle_links()
+
+ assert seen == {"run_id": RUN, "request": ""} # inside the run, nothing counts as a request
+ assert run_id_var.get() == "" # the caller's context is untouched
+ assert _links(sessions) == [("start", "trace-start")]
+
+
+@pytest.mark.asyncio
+async def test_run_started_without_a_request_links_nothing(sessions):
+ async def worker():
+ return None
+
+ runtime.register_run_task(RUN, worker(), name=f"test:{RUN}") # e.g. startup recovery
+ await runtime.tasks.process_state().active_runs[RUN]
+ await _settle_links()
+ assert _links(sessions) == []
+
+
+@pytest.mark.asyncio
+async def test_answer_links_the_answering_request_to_the_run_that_asked(sessions):
+ async def ask():
+ await interaction_store.register("question", "chat-links", "q1", {"questions": []}, 60)
+
+ run_token = run_id_var.set(RUN)
+ try:
+ await asyncio.create_task(ask()) # registered from inside the run's worker task
+ finally:
+ run_id_var.reset(run_token)
+
+ trace_token = trace_id_var.set("trace-answer")
+ try:
+ assert await interaction_store.decide("question", "chat-links", "q1", {"outcome": "answered"})
+ finally:
+ trace_id_var.reset(trace_token)
+ await interaction_store.remove("question", "chat-links", "q1")
+
+ assert _links(sessions) == [("answer:question", "trace-answer")]
+
+
+@pytest.mark.asyncio
+async def test_actions_taken_inside_a_run_are_not_linked(sessions):
+ run_token = run_id_var.set("some-other-run")
+ trace_token = trace_id_var.set("inherited-trace")
+ try:
+ await request_links.link_request(RUN, "cancel")
+ finally:
+ trace_id_var.reset(trace_token)
+ run_id_var.reset(run_token)
+ assert _links(sessions) == []
+
+
+def test_log_lines_carry_the_run_id_without_overriding_an_explicit_one():
+ token = run_id_var.set(RUN)
+ try:
+ assert add_user_context(None, "info", {"event": "x"})["run_id"] == RUN
+ assert add_user_context(None, "info", {"event": "x", "run_id": "other"})["run_id"] == "other"
+ finally:
+ run_id_var.reset(token)
diff --git a/src/backend/tests/core/runs/test_task_numbers.py b/src/backend/tests/core/runs/test_task_numbers.py
new file mode 100644
index 00000000..98e77e36
--- /dev/null
+++ b/src/backend/tests/core/runs/test_task_numbers.py
@@ -0,0 +1,153 @@
+"""Task numbers: RW-YYYYMMDD-NNNN per run, RW-…-NN per sub-agent call, backfilled on upgrade."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import core.harness.log_records as log_records
+import pytest
+from core.db.engine import Base
+from core.db.models import ChatRun, ChatSession, SubAgentCallLog, TaskNumberCounter
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+NOON = datetime(2026, 10, 8, 4, 0, tzinfo=timezone.utc)
+
+
+def _day(at: datetime) -> str:
+ return f"{at.astimezone():%Y%m%d}"
+
+
+@pytest.fixture()
+def factory(tmp_path, monkeypatch):
+ engine = create_engine(
+ f"sqlite:///{tmp_path / 'task-numbers.sqlite'}", connect_args={"check_same_thread": False}
+ )
+ Base.metadata.create_all(engine)
+ sessions = sessionmaker(bind=engine, expire_on_commit=False)
+ with sessions() as db:
+ db.add(ChatSession(chat_id="chat-1", user_id="u1", title="t"))
+ db.commit()
+ monkeypatch.setattr(log_records, "SessionLocal", sessions)
+ monkeypatch.setattr(log_records.log_writer_constants, "SUBAGENT_LOG_ENABLED", True)
+ yield sessions
+ engine.dispose()
+
+
+def _add_run(sessions, run_id: str, at: datetime = NOON) -> str:
+ with sessions() as db:
+ run = ChatRun(run_id=run_id, chat_id="chat-1", user_id="u1", message_id=f"msg-{run_id}", created_at=at)
+ db.add(run)
+ db.commit()
+ return run.task_no
+
+
+def _sub(sessions, log_id: str, run_id: str, *, kind: str = "user_agent", parent: str | None = None) -> str:
+ log_records.write_subagent_create_sync({
+ "id": log_id, "run_id": run_id, "subagent_name": log_id, "subagent_type": kind,
+ "parent_subagent_log_id": parent, "message_id": f"msg-{run_id}",
+ })
+ with sessions() as db:
+ return db.get(SubAgentCallLog, log_id).task_no
+
+
+def test_runs_are_numbered_per_day_in_order(factory):
+ next_day = NOON + timedelta(days=1)
+ numbers = [_add_run(factory, "r1"), _add_run(factory, "r2"), _add_run(factory, "r3", next_day)]
+ assert numbers == [f"RW-{_day(NOON)}-0001", f"RW-{_day(NOON)}-0002", f"RW-{_day(next_day)}-0001"]
+
+
+def test_concurrent_admissions_never_share_a_number(factory):
+ with ThreadPoolExecutor(max_workers=8) as pool:
+ numbers = list(pool.map(lambda i: _add_run(factory, f"c{i}"), range(24)))
+ assert sorted(numbers) == [f"RW-{_day(NOON)}-{n:04d}" for n in range(1, 25)]
+ with factory() as db:
+ assert db.get(TaskNumberCounter, _day(NOON)).last_seq == 24
+
+
+def test_subtasks_are_numbered_under_whoever_dispatched_them(factory):
+ main = _add_run(factory, "r1")
+ assert _sub(factory, "a", "r1") == f"{main}-01"
+ assert _sub(factory, "b", "r1") == f"{main}-02"
+ assert _sub(factory, "a1", "r1", parent="a") == f"{main}-01-01" # called by sub-agent a
+ assert _sub(factory, "c", "r1") == f"{main}-03"
+
+
+def test_plan_steps_number_under_the_run_and_the_coordinator_keeps_its_number(factory):
+ main = _add_run(factory, "r1")
+ assert _sub(factory, "plan", "r1", kind="plan_mode") == main
+ assert _sub(factory, "s1", "r1", kind="plan_step", parent="plan") == f"{main}-01"
+ assert _sub(factory, "s2", "r1", kind="plan_step", parent="plan") == f"{main}-02"
+
+
+def test_subagent_outside_any_run_has_no_number(factory):
+ assert _sub(factory, "x", "") is None
+
+
+# ── Upgrade backfill ───────────────────────────────────────────────────────
+
+
+def _alembic(repo_root: Path, url: str, *args: str) -> None:
+ result = subprocess.run(
+ [sys.executable, "-m", "alembic", *args], cwd=repo_root,
+ env={**os.environ, "DATABASE_URL": url}, capture_output=True, text=True, check=False,
+ )
+ assert result.returncode == 0, result.stdout + result.stderr
+
+
+def test_upgrade_numbers_existing_runs_and_subagent_calls(tmp_path):
+ repo_root = Path(__file__).resolve().parents[5]
+ revision, previous = (
+ ("ce_0036", "ce_0035")
+ if (repo_root / ".hugagent-edition").is_file()
+ else ("tasknumber01", "taskmonitor02")
+ )
+ url = f"sqlite:///{tmp_path / 'backfill.sqlite'}"
+ engine = create_engine(url)
+ # The full chain does not run from scratch on SQLite; start from the current schema and
+ # step back over this revision only.
+ Base.metadata.create_all(engine)
+ _alembic(repo_root, url, "stamp", revision)
+ _alembic(repo_root, url, "downgrade", previous)
+ day1, day2 = NOON, NOON + timedelta(days=1)
+ with engine.begin() as db:
+ db.execute(text(
+ "INSERT INTO chat_sessions (chat_id, user_id, title, created_at, updated_at) "
+ "VALUES ('c', 'u', 't', :at, :at)"
+ ), {"at": day1})
+ for run_id, at in (("late", day2), ("first", day1), ("second", day1 + timedelta(minutes=1))):
+ db.execute(text(
+ "INSERT INTO chat_runs (run_id, chat_id, user_id, message_id, status, last_event_offset, "
+ "run_phase, operation_seq, snapshot_version, last_operation_safety, created_at, updated_at) "
+ "VALUES (:r, 'c', 'u', :m, 'completed', 0, 'completed', 0, 0, 'replayable', :at, :at)"
+ ), {"r": run_id, "m": f"m-{run_id}", "at": at})
+ for i, (log_id, kind, parent) in enumerate((
+ ("plan", "plan_mode", None), ("s1", "plan_step", "plan"), ("s2", "plan_step", "plan"),
+ ("helper", "user_agent", "s1"),
+ )):
+ db.execute(text(
+ "INSERT INTO subagent_call_logs (id, message_id, subagent_name, subagent_type, "
+ "parent_subagent_log_id, status, created_at) VALUES (:i, 'm-first', :i, :k, :p, 'success', :at)"
+ ), {"i": log_id, "k": kind, "p": parent, "at": day1 + timedelta(seconds=i)})
+ _alembic(repo_root, url, "upgrade", revision)
+
+ with engine.connect() as db:
+ runs = dict(db.execute(text("SELECT run_id, task_no FROM chat_runs")).all())
+ logs = dict(db.execute(text("SELECT id, task_no FROM subagent_call_logs")).all())
+ counters = dict(db.execute(text("SELECT day, last_seq FROM task_number_counters")).all())
+ first = f"RW-{_day(day1)}-0001"
+ assert runs == {"first": first, "second": f"RW-{_day(day1)}-0002", "late": f"RW-{_day(day2)}-0001"}
+ assert logs == {"plan": first, "s1": f"{first}-01", "s2": f"{first}-02", "helper": f"{first}-01-01"}
+ assert counters == {_day(day1): 2, _day(day2): 1}
+
+ with sessionmaker(bind=engine)() as db: # new runs continue the day's sequence
+ db.add(ChatRun(run_id="new", chat_id="c", user_id="u", message_id="m-new", created_at=day1))
+ db.commit()
+ assert db.get(ChatRun, "new").task_no == f"RW-{_day(day1)}-0003"
+ _alembic(repo_root, url, "downgrade", previous)
+ engine.dispose()
diff --git a/src/backend/tests/llm/test_context_step_replay.py b/src/backend/tests/llm/test_context_step_replay.py
new file mode 100644
index 00000000..5f236ff4
--- /dev/null
+++ b/src/backend/tests/llm/test_context_step_replay.py
@@ -0,0 +1,159 @@
+"""Hook-injected context is part of the model's own history.
+
+Attachment previews, the historical-file listing and reasoning reminders enter
+the live context through hooks. The next turn rebuilds history from the stored
+rows; if those blocks are missing there, the request stops matching the one the
+provider cached at the first such block and everything after it is billed again.
+These tests pin the live request and the replayed one to the same bytes.
+"""
+
+import json
+from types import SimpleNamespace
+
+import pytest
+from agentscope.message import Msg, TextBlock, ThinkingBlock, ToolCallBlock, ToolResultBlock
+
+from core.chat.compaction.history import normalize_rows
+from core.llm.context.ir_constants import KIND_USER_INPUT, POLICY_NEVER
+from core.llm.context.items import ContextItem
+from core.llm.context.rendering import next_request_sequence, render_context_item
+from core.llm.runtime.context_hook_adapter import append_hook_context, record_hook_context
+from core.llm.runtime.message_compat import session_to_msgs
+from core.llm.runtime.model_steps import record_assistant_step, record_tool_result_step
+from core.models.chat_models import ReasoningEchoChatFormatter
+
+ATTACHMENT = "[file name]: - sales.csv (file_id: ua_1)\n[file content begin]\n| 区域 | 销售额 |\n[file content end]"
+
+
+@pytest.fixture
+def media_db(tmp_path, monkeypatch):
+ from core.db.models import ToolMediaBlob
+ from core.llm.runtime import tool_media_store
+ from sqlalchemy import create_engine
+ from sqlalchemy.orm import sessionmaker
+
+ engine = create_engine(f"sqlite:///{tmp_path}/media.db")
+ ToolMediaBlob.__table__.create(engine, checkfirst=True)
+ monkeypatch.setattr(tool_media_store, "SessionLocal", sessionmaker(bind=engine))
+ tool_media_store.reset_cache()
+ yield engine
+ tool_media_store.reset_cache()
+
+
+def _agent(context=()):
+ return SimpleNamespace(state=SimpleNamespace(context=list(context), pending_model_steps=[]))
+
+
+def _item(content, kind="attachment", origin="user:uploaded_files", trust="user"):
+ return dict(content=content, kind=kind, origin=origin, trust=trust,
+ priority=650, token_budget=8000)
+
+
+async def _inject(agent, *items, before_input):
+ appended = append_hook_context(agent, SimpleNamespace(data={"context_items": list(items)}))
+ await record_hook_context(agent, appended, before_input=before_input)
+
+
+def _request(context, text):
+ seq = next_request_sequence(context)
+ return render_context_item(ContextItem.create(
+ item_id=f"request:{seq}", kind=KIND_USER_INPUT, origin="user:chat", trust="user",
+ visibility="model", priority=1_000, token_budget=100_000,
+ truncation_policy=POLICY_NEVER, content=text, cache_class="dynamic",
+ created_seq=seq, render_role="user", render_name="user",
+ message_group=f"request:{seq}",
+ ))
+
+
+def _assistant(*blocks):
+ return record_assistant_step(blocks, provider="deepseek", model="v4", protocol="openai_chat")
+
+
+def _row(role, content, model_steps=None, **extra):
+ return SimpleNamespace(role=role, content=content, extra_data=extra, model_steps=model_steps)
+
+
+async def _wire(messages):
+ return await ReasoningEchoChatFormatter().format(messages)
+
+
+def _drain(agent):
+ steps, agent.state.pending_model_steps = agent.state.pending_model_steps, []
+ return steps
+
+
+@pytest.mark.asyncio
+async def test_attachment_turn_replays_to_the_request_the_model_was_sent():
+ history = [Msg(name="user", role="user", content=[TextBlock(type="text", text="上一问")]),
+ Msg(name="assistant", role="assistant", content=[TextBlock(type="text", text="上一答")])]
+ agent = _agent(history)
+ # on_reply: the attachment preview is appended before AgentScope adds the user's message.
+ await _inject(agent, _item(ATTACHMENT), before_input=True)
+ agent.state.context.append(_request(agent.state.context, "按区域汇总"))
+ answer = TextBlock(type="text", text="汇总完成")
+ agent.state.pending_model_steps.append(_assistant(answer))
+ agent.state.context.append(Msg(name="assistant", role="assistant", content=[answer]))
+
+ replayed = normalize_rows([
+ _row("user", "上一问"),
+ _row("assistant", "上一答"),
+ _row("user", "按区域汇总"),
+ _row("assistant", "汇总完成", model_steps=_drain(agent)),
+ ])
+
+ assert await _wire(session_to_msgs(replayed)) == await _wire(agent.state.context)
+
+
+@pytest.mark.asyncio
+async def test_reasoning_reminder_replays_between_the_steps_it_was_injected_into():
+ agent = _agent()
+ agent.state.context.append(_request(agent.state.context, "查一下"))
+ call = ToolCallBlock(type="tool_call", id="c1", name="search", input="{}")
+ result = ToolResultBlock(type="tool_result", id="c1", name="search", output="结果")
+ agent.state.context.append(Msg(name="assistant", role="assistant",
+ content=[ThinkingBlock(type="thinking", thinking="先查"), call, result]))
+ steps = [_assistant(ThinkingBlock(type="thinking", thinking="先查"), call),
+ record_tool_result_step(result)]
+ # on_reasoning: a reminder lands after the tool result, before the next model call.
+ await _inject(agent, _item("