[Bug]: Unify task tracking and same-session concurrency semantics

[rayrayraykk]

QwenPaw Version

QwenPaw 2.0.1b1, reproduced from source near commit 8c2cceff.

Description

Task tracking and same-session concurrency behave differently depending on
the execution entry point. Some paths serialize work, some attach to an
existing run while silently ignoring the new payload, and others allow
concurrent execution without registering the real task in TaskTracker.

This makes chat status, cancellation, graceful workspace reload, and session
state safety depend on how the run was started rather than on a documented
policy.

Examples:

• External channel messages with the same (channel, session, priority) are
  serialized by UnifiedQueueManager, while different priorities may run
  concurrently for the same session.
• A second /api/console/chat request for an active chat attaches to the
  existing tracked run. Its new payload is not executed, and the API does not
  report a conflict or queued status.
• /api/console/chat/task bypasses both UnifiedQueueManager and
  TaskTracker, so multiple background tasks using the same session can run
  concurrently.
• Cron agent runs create a ChatSpec, but max_concurrency can permit
  multiple executions that reuse the same job session.
• Heartbeat runs use the shared main session and can be triggered manually
  in addition to scheduled execution, but they do not explicitly create a
  ChatSpec or register the real execution in TaskTracker.
• ACP maintains its own session registry and cancellation events. Concurrent
  prompts for one ACP session are not explicitly rejected or serialized, and
  a later prompt can replace the earlier _cancel_events[session_id] entry.

Related PR(s): #6272

Security considerations: No direct data-exposure issue is known. However,
concurrent writes and incomplete cancellation can make tool execution and
session state harder to reason about.

Component(s) Affected

• Core / Backend (app, agents, config, providers, utils, local_models)
• Console (frontend web UI)
• Channels (DingTalk, Feishu, QQ, Discord, iMessage, etc.)
• Skills
• CLI
• Documentation (website)
• Tests
• CI/CD
• Scripts / Deploy

Environment

• QwenPaw version: 2.0.1b1 / source commit 8c2cceff
• OS: macOS
• Install method: From source
• Python version (if applicable): Python 3.12, conda environment

Steps to Reproduce

1. Submit two POST /api/console/chat/task requests with the same
   session_id before either task completes.
2. Observe that both tasks receive distinct task IDs and execute concurrently.
3. Query the chat list while they run. The chat can exist, but its status is
   not backed by the actual background tasks in TaskTracker.
4. Submit two regular POST /api/console/chat requests for the same active
   chat.
5. Observe that the second request attaches to the first run instead of
   executing, queueing, or returning an explicit conflict.
6. Trigger /api/config/heartbeat/run while another main session run is
   active, or trigger it repeatedly, and observe that no common per-session
   concurrency policy is applied.

Actual vs Expected

• Actual: Execution identity is sometimes a chat ID, sometimes an opaque
  task ID, and sometimes absent from TaskTracker. Same-session work may be
  serialized, merged, silently ignored, attached, or run concurrently based
  on the entry point. request_stop() cannot cancel tasks that were never
  registered with their actual asyncio.Task.
• Expected: All execution paths should follow an explicit and testable
  same-session concurrency policy. Every running operation should have a
  unique execution ID, an optional chat association, accurate aggregate chat
  status, and cancellation tied to the real task.

Logs / Screenshots

Relevant code paths:

src/qwenpaw/app/channels/unified_queue_manager.py
  QueueKey = (channel_id, session_id, priority_level)

src/qwenpaw/app/channels/base.py
  TaskTracker.attach_or_start(chat.id, ...)
  Logs "Message ignored (task already running)" for an existing run.

src/qwenpaw/app/routers/console.py
  /console/chat uses TaskTracker.attach_or_start(chat.id, ...)
  /console/chat/task creates an independent asyncio task outside TaskTracker.

src/qwenpaw/app/crons/executor.py
  Cron agent runs create or reuse a ChatSpec.

src/qwenpaw/app/crons/heartbeat.py
  Heartbeat runs use session_id="main" and call workspace.stream_query().

src/qwenpaw/agents/acp/server.py
  ACP prompt stores one cancel event per session ID and calls
  workspace.stream_query() directly.

Additional Notes

Suggested design direction:

1. Give every execution a unique run_id or task_id.
2. Associate executions with an optional chat_id instead of using
   chat_id as the only TaskTracker key.
3. Allow TaskTracker to aggregate multiple running executions for one chat.
4. Register the real asyncio.Task so stop and shutdown operations cancel or
   await the actual execution.
5. Define one policy for duplicate same-session work: queue, reject with an
   explicit conflict, merge under documented rules, or allow controlled
   concurrency. Avoid silently dropping the new payload.
6. Apply the policy consistently to Console streaming, background chat tasks,
   external channels, cron, heartbeat, and ACP.

Suggested acceptance criteria:

• Two runs for the same session have deterministic behavior documented by the
  selected policy.
• Chat status remains running until all associated executions finish.
• Stop/cancel targets real executions and does not only update proxy state.
• Workspace reload waits for or cancels all registered executions.
• Integration tests cover regular Console chat, background chat tasks,
  cross-priority channel messages, heartbeat, cron concurrency, and ACP
  prompts.

[github-actions]

Welcome to QwenPaw! 🐾

Hi @rayrayraykk, this is your 17th issue.

We'll review your issue soon. Thank you for supporting QwenPaw!

[wananing]

Thanks for documenting this. I traced these paths on current main after #6272, and I agree that this should be treated as a shared execution contract rather than a set of entry-point-specific guards.

My suggested model is:

• Give every execution a unique run_id, associate it with the real persistence key (channel, user_id, session_id) and an optional chat_id, and register the real asyncio.Task.
• Separate execution lifecycle/admission from SSE subscription and replay. Reconnection should be an explicit attach operation, not an implicit result of submitting a second payload.
• Enforce at most one state-mutating execution per persistence key by default, while allowing control-plane commands such as /stop and /status to run out of band and target the registry.
• Aggregate chat status over associated executions, and make stop/reload wait for or cancel the actual registered tasks.

A possible policy matrix would be:

Entry point	Same-session conflict
Regular Console chat	return an explicit conflict for a new payload; attach only when reconnect=true
Background Console task	allocate a task ID immediately, expose queued, then execute serially
External channels	retain normal-message queueing; apply the session gate across non-control priorities
Cron	serialize jobs sharing a session; permit max_concurrency > 1 only with isolated per-run sessions
Heartbeat	coalesce or skip a duplicate run while its shared session is active
ACP	explicitly reject a concurrent prompt for the same session, matching the existing ACP client turn lock behavior

The priority distinction matters: putting one lock around every channel priority would also delay /stop, defeating cancellation. The common invariant should apply to state-mutating runs, not control commands.

This direction also addresses the underlying cron case in #4217 and gives an explicit successor policy for the Console behavior discussed in #5344. The atomic writes added by #4706 prevent torn JSON files, but they do not prevent two runtimes from loading the same old state and then overwriting each other logically.

Would this policy direction be acceptable? If so, would maintainers prefer one larger PR, or staged PRs with a compatibility layer (registry plus Console/background/channels first, then cron/heartbeat/ACP)?

[chrischen-coder]

I am working on the focused Console admission failure described here: a second non-reconnect POST /api/console/chat currently attaches to the active run and silently discards its new payload.

I will keep reconnect behavior explicit (reconnect=true), return a machine-visible HTTP 409 for a conflicting new payload, and add a regression proving the existing run continues without leaking an extra subscriber or invoking the second payload. This is intentionally scoped to the regular Console endpoint rather than claiming to solve the full cross-entry-point execution contract.

[chrischen-coder]

Focused fix submitted as #7299.

It makes regular Console admission fail closed: a second non-reconnect payload for an active chat now receives a structured HTTP 409 instead of HTTP 200 on a replay stream while its payload is silently ignored. Explicit reconnect=true behavior is unchanged.

The regression holds the original run open and verifies that the second producer is never invoked, the original run continues, and the temporary replay subscriber is detached. This intentionally addresses only the regular Console endpoint portion of the broader execution-contract issue.