Refactor/channel system

[alexhoshina]

📝 Description

Comprehensive refactoring of the channel system (pkg/channels/), message bus (pkg/bus/), and supporting infrastructure across 12 phases. This PR restructures how PicoClaw manages chat platform connections — from a flat, tightly-coupled package into a modular, capability-driven architecture with proper concurrency control, error handling, and lifecycle management.

Impact: ~5,900 additions / ~1,400 deletions across 68 files, touching all 12 channel implementations + bus + agent loop + config + 3 new packages.

🗣️ Type of Change

• 🐞 Bug fix (non-breaking change which fixes an issue)
• ✨ New feature (non-breaking change which adds functionality)
• 📖 Documentation update
• ⚡ Code refactoring (functional changes, api changes)

🔗 Related Issue

Closes #621

Resolves: #310 #381 #619 #449 #244 #285 #216 #9 #36

📚 Technical Context

Why This Refactoring

The channel system had accumulated significant technical debt: implicit metadata contracts (peer_kind, peer_id as magic strings), media files deleted before async consumers read them, a single dispatcher goroutine where one slow channel blocks all others, no retry/rate-limiting on sends, scattered group-trigger logic duplicated across channels, and a MessageBus that could deadlock when the buffer was full during shutdown.

This PR addresses all of these systematically across 12 phases.

---

Architecture Changes

1. Channel Subpackages + Factory Registry

Channels moved from flat files (pkg/channels/telegram.go) into subpackages (pkg/channels/telegram/telegram.go). Each registers itself via init() + RegisterFactory() — the same pattern as database/sql drivers. The Manager dynamically creates channels from config without import-time coupling.

pkg/channels/
├── base.go              # Enhanced BaseChannel with group triggers, media store, options
├── manager.go           # Worker queues, rate limiting, retry, typing orchestration
├── registry.go          # Factory registry (RegisterFactory / CreateChannel)
├── interfaces.go        # TypingCapable, MessageEditor, PlaceholderRecorder
├── webhook.go           # WebhookHandler, HealthChecker interfaces
├── errors.go            # ErrNotRunning, ErrRateLimit, ErrTemporary, ErrSendFailed
├── errutil.go           # ClassifySendError, ClassifyNetError helpers
├── split.go             # Rune-aware message splitting (moved from pkg/utils)
├── media.go             # MediaSender interface
├── telegram/            # Per-channel subpackages...
├── discord/
├── slack/
├── onebot/
├── line/
├── qq/
├── dingtalk/
├── feishu/
├── wecom/
├── whatsapp/
├── maixcam/
└── pico/                # NEW: Native WebSocket Pico Protocol

2. MessageBus Concurrency Fix

Replaced sync.RWMutex + bool with atomic.Bool + done channel. Publish methods use a lock-free 3-way select (send | done | ctx.Done), eliminating the deadlock where PublishInbound blocked on a full buffer while holding the read lock, preventing Close() from acquiring the write lock.

• All publish/consume methods now accept context.Context
• Close() drains buffered messages instead of silently dropping them
• Removed dead code: RegisterHandler, GetHandler, handlers map
• Added ErrBusClosed sentinel error
• 11 test cases covering round-trip, cancellation, concurrent publish+close, full-buffer timeout

3. Structured Identity (bus.Peer + pkg/identity)

Promoted peer_kind/peer_id/message_id from implicit metadata strings to typed struct fields on InboundMessage. New pkg/identity package provides canonical platform:id format and backward-compatible allow-list matching (supports legacy numeric, @username, id|username, and new telegram:123456 formats).

4. MediaStore File Lifecycle (pkg/media)

Channels previously deleted downloaded media via defer os.Remove, racing with the async Agent consumer. FileMediaStore decouples file ownership: channels register files on download (Store() → media://<uuid> ref), Agent resolves refs after processing. Scope-based ReleaseAll() implemented but currently disabled pending proper consumption confirmation.

5. Per-Channel Worker Queues + Rate Limiting

Manager now runs a dedicated worker goroutine per channel with buffered queues, replacing the single dispatcher that could block all channels. Each worker has a golang.org/x/time/rate token-bucket limiter with platform-specific defaults (Discord 1 msg/s burst 5, Telegram 20 msg/s, etc.).

6. Error Classification + Retry

Defined sentinel errors (ErrNotRunning, ErrRateLimit, ErrTemporary, ErrSendFailed). All 12 channels return typed errors from Send(). Manager's sendWithRetry() classifies errors and applies the right strategy: permanent → bail, rate-limit → fixed 1s delay, temporary → exponential backoff (500ms→1s→2s, max 3 retries).

7. Unified Channel Lifecycle

All channels follow a standard contract:

• Start(): create cancelCtx via context.WithCancel, spawn goroutines, SetRunning(true)
• Stop(): call cancel(), wait for goroutines, release resources, SetRunning(false)
• Send(): check IsRunning(), respect ctx.Done(), return typed errors

Fixed goroutine/context leaks in OneBot, Telegram, Discord, WhatsApp, and MaixCam.

8. Shared HTTP Server

Merged 3 independent webhook servers (LINE :18791, WeCom Bot :18793, WeCom App :18792) + health server (:18790) into a single shared HTTP server on the Gateway address. Channels implement WebhookHandler and/or HealthChecker interfaces to register routes. Default listen address changed from 0.0.0.0 to 127.0.0.1 for security.

9. Outbound Media via Bus

New MediaSender optional interface + OutboundMediaMessage bus type. Agent publishes media from tool results via bus; Manager dispatches to channels that implement SendMedia(). Implemented for 7 channels: Telegram, Discord, Slack, OneBot, LINE, WeCom App, Pico.

10. Group Trigger Standardization

Unified ShouldRespondInGroup(isMentioned, content) in BaseChannel replaces 9 scattered per-channel implementations. Configurable via GroupTriggerConfig with mention_only and prefixes fields. Legacy config fields (Discord.MentionOnly, OneBot.GroupTriggerPrefix) auto-migrate.

11. Typing/Placeholder Orchestration

Manager tracks typing indicators and placeholder messages via PlaceholderRecorder interface. On outbound message, preSend() stops typing and attempts to edit the placeholder (if channel implements MessageEditor). Implemented for Telegram, Discord, and Pico.

12. Pico Protocol Channel (New)

Native WebSocket channel as a reference implementation of all optional interfaces (Channel, MessageEditor, TypingCapable, MediaSender, WebhookHandler). JSON protocol with message.send/create/update, typing.start/stop, media.send/create, ping/pong. Bearer token auth with configurable query-param fallback.

---

Message Flow (After Refactoring)

Inbound:
  Platform → Channel.Start (poll/webhook/WS)
  → Parse + download media → MediaStore.Store(scope)
  → [Optional] Channel triggers typing/placeholder via Manager
  → BaseChannel.HandleMessage → bus.PublishInbound(ctx)
  → Agent.processMessage

Outbound:
  Agent done → bus.PublishOutbound / PublishOutboundMedia
  → Manager.dispatchOutbound → per-channel worker queue
  → preSend: stop typing + edit placeholder (if MessageEditor)
  → SplitMessage (rune-aware, Markdown-preserving)
  → Rate limit (token bucket)
  → sendWithRetry (classify errors, backoff)
  → channel.Send()

New Packages

Package	Purpose
pkg/media	MediaStore interface + FileMediaStore for file lifecycle with scope-based cleanup
pkg/identity	Canonical platform:id format, MatchAllowed() with backward-compatible allow-list matching
pkg/channels/pico	Native WebSocket Pico Protocol channel (reference implementation)

Test Coverage

Added 1,816 lines of new test code across 12 test files:

Test File	Cases	Coverage
bus/bus_test.go	11	Round-trip, context cancellation, closed-bus, concurrent publish+close, buffer-full timeout, idempotent close
channels/manager_test.go	21	Retry logic (temp/perm/rate-limit/max-retries/backoff timing), worker rate limiting, message splitting, preSend placeholder editing, typing-stop, concurrency safety
channels/base_test.go	24	Allow-list matching (5 formats), group trigger (12 scenarios), IsAllowedSender (7 cases)
channels/errors_test.go	9	Sentinel error wrapping, errors.Is semantics, error messages
channels/errutil_test.go	11	HTTP status→error classification, false positive guards, net error wrapping
channels/split_test.go	8	Empty/short/long messages, newline splitting, code block integrity, rune-aware Unicode
media/store_test.go	7	Store/resolve, scope isolation, ReleaseAll idempotence, error propagation, metadata, concurrent safety
identity/identity_test.go	36	Canonical ID build/parse, MatchAllowed (17 cases covering all legacy formats), isNumeric
routing/session_key_test.go	24	Session key build/parse, DM scope modes, identity link resolution

Known Limitations

1. Media cleanup disabled: ReleaseAll() is commented out in agent loop to prevent premature file deletion. Files in /tmp/picoclaw_media/ will accumulate. A TTL-based or periodic cleanup strategy is needed.
2. Voice transcription removed from channels but Agent-layer middleware not yet implemented — audio messages are currently stored but not transcribed.
3. Rate limiter values hardcoded with conservative defaults (Discord 1 msg/s, Slack 1 msg/s). Should be configurable or adjusted.
4. MediaSender not implemented for QQ, DingTalk, Feishu, WhatsApp.

☑️ Checklist

• My code/docs follow the style of this project.
• I have performed a self-review of my own changes.
• I have updated the documentation accordingly.

[nikolasdehor]

Substantial refactor — the channel system is significantly better structured after this. Key improvements: self-registering channel factories, context-aware bus with proper close semantics, structured sender identity, media lifecycle store, rate-limited per-channel workers, and unified group trigger logic.

Reviewed the full diff. Several observations:

Architecture (positive)

• The init() + RegisterFactory pattern for channels is clean and idiomatic Go (similar to database/sql drivers). Blank imports in cmd_gateway.go make the dependencies explicit.
• MessageBus.Close() using atomic.Bool + a done channel instead of sync.RWMutex eliminates the potential deadlock the old code had when PublishInbound blocked on a full buffer while holding the read lock.
• The SplitMessage logic move from pkg/utils/message.go into pkg/channels/split.go is appropriate — message splitting is channel-specific.
• FileMediaStore with scope-based lifecycle is a good abstraction for cross-channel media.

Potential issues

1. Bus channels never closed: NewMessageBus creates buffered channels (inbound, outbound, outboundMedia) but Close() only closes done — it never closes the actual channels. This means messages buffered in those channels at shutdown are silently dropped. The old code closed both channels. Consider draining or closing them in Close() (after setting the closed flag).

2. Rate limiter values hardcoded: channelRateConfig has Discord and Slack at 1 msg/s. Discord's actual limit is 5 msg/s per channel, Slack is 1 msg/s per method. These should probably be configurable or at least documented that they are conservative defaults.

3. context.TODO() usage: Several call sites use context.TODO() for PublishOutbound/PublishInbound (e.g., cron.go, subagent.go, summarize goroutine). These would silently hang if the bus buffer is full and the bus is not yet closed. Consider using a context with timeout for fire-and-forget publishes.

4. ShouldRespondInGroup prefix matching: The current implementation uses strings.HasPrefix which is case-sensitive. If a user types /Ask instead of /ask, it won't match. This may be intentional, but worth documenting.

5. Shutdown order change: The old code called channelManager.StopAll(ctx) last; the new code calls it first (before stopping cron, heartbeat, agent). This seems correct (stop accepting new messages first), but the shutdownCtx with 15s timeout is shared across all the stop calls that follow. If StopAll takes close to 15s, the subsequent stops get no grace period. Might be fine in practice but worth noting.

6. Media cleanup disabled: The TODO comment in loop.go notes media cleanup is disabled because files are deleted before the LLM accesses them. This means /tmp/picoclaw_media/ files will accumulate indefinitely. Consider a periodic cleanup or TTL-based sweep in FileMediaStore.

7. OneBotChannel.Stop — pending channels not closed: In the old code, close(ch) was called for pending response channels. The new code does delete(c.pending, echo) without closing. Any goroutine blocked on a read from that channel will now leak. This looks like a bug.

8. uuid dependency: The new google/uuid dependency is only used in BuildMediaScope and FileMediaStore.Store. Both usages are fine, but the import in base.go means the channels package itself now depends on uuid.

Test coverage: Good additions — bus_test.go, base_test.go, identity_test.go, media/store_test.go, manager_test.go, session_key_test.go, errutil_test.go all cover the new abstractions well. The concurrent safety test for FileMediaStore is appreciated.

Overall this is solid work. The items above are mostly about edge cases around shutdown and resource cleanup. Item 7 (OneBot pending channel leak) looks like an actual bug that should be addressed.

[alexhoshina]

Substantial refactor — the channel system is significantly better structured after this. Key improvements: self-registering channel factories, context-aware bus with proper close semantics, structured sender identity, media lifecycle store, rate-limited per-channel workers, and unified group trigger logic.

Reviewed the full diff. Several observations:

Architecture (positive)

• The init() + RegisterFactory pattern for channels is clean and idiomatic Go (similar to database/sql drivers). Blank imports in cmd_gateway.go make the dependencies explicit.
• MessageBus.Close() using atomic.Bool + a done channel instead of sync.RWMutex eliminates the potential deadlock the old code had when PublishInbound blocked on a full buffer while holding the read lock.
• The SplitMessage logic move from pkg/utils/message.go into pkg/channels/split.go is appropriate — message splitting is channel-specific.
• FileMediaStore with scope-based lifecycle is a good abstraction for cross-channel media.

Potential issues

1. Bus channels never closed: NewMessageBus creates buffered channels (inbound, outbound, outboundMedia) but Close() only closes done — it never closes the actual channels. This means messages buffered in those channels at shutdown are silently dropped. The old code closed both channels. Consider draining or closing them in Close() (after setting the closed flag).
2. Rate limiter values hardcoded: channelRateConfig has Discord and Slack at 1 msg/s. Discord's actual limit is 5 msg/s per channel, Slack is 1 msg/s per method. These should probably be configurable or at least documented that they are conservative defaults.
3. context.TODO() usage: Several call sites use context.TODO() for PublishOutbound/PublishInbound (e.g., cron.go, subagent.go, summarize goroutine). These would silently hang if the bus buffer is full and the bus is not yet closed. Consider using a context with timeout for fire-and-forget publishes.
4. ShouldRespondInGroup prefix matching: The current implementation uses strings.HasPrefix which is case-sensitive. If a user types /Ask instead of /ask, it won't match. This may be intentional, but worth documenting.
5. Shutdown order change: The old code called channelManager.StopAll(ctx) last; the new code calls it first (before stopping cron, heartbeat, agent). This seems correct (stop accepting new messages first), but the shutdownCtx with 15s timeout is shared across all the stop calls that follow. If StopAll takes close to 15s, the subsequent stops get no grace period. Might be fine in practice but worth noting.
6. Media cleanup disabled: The TODO comment in loop.go notes media cleanup is disabled because files are deleted before the LLM accesses them. This means /tmp/picoclaw_media/ files will accumulate indefinitely. Consider a periodic cleanup or TTL-based sweep in FileMediaStore.
7. OneBotChannel.Stop — pending channels not closed: In the old code, close(ch) was called for pending response channels. The new code does delete(c.pending, echo) without closing. Any goroutine blocked on a read from that channel will now leak. This looks like a bug.
8. uuid dependency: The new google/uuid dependency is only used in BuildMediaScope and FileMediaStore.Store. Both usages are fine, but the import in base.go means the channels package itself now depends on uuid.

Test coverage: Good additions — bus_test.go, base_test.go, identity_test.go, media/store_test.go, manager_test.go, session_key_test.go, errutil_test.go all cover the new abstractions well. The concurrent safety test for FileMediaStore is appreciated.

Overall this is solid work. The items above are mostly about edge cases around shutdown and resource cleanup. Item 7 (OneBot pending channel leak) looks like an actual bug that should be addressed.

Thank you for your review, I will make the necessary fixes later

[nikolasdehor]

@alexhoshina Sounds good, take your time with the fixes. For prioritization, I'd focus on:

1. Item 7 (OneBot pending channel leak) — this is a correctness bug that will cause goroutine leaks in production. Should be straightforward to add close(ch) before delete(c.pending, echo).

2. Item 1 (bus channels not closed on shutdown) — buffered messages silently dropped. Draining after setting the closed flag would preserve them.

3. Item 3 (context.TODO() in fire-and-forget publishes) — a context with 5-10s timeout prevents silent hangs.

The rest (rate limiter values, ShouldRespondInGroup case sensitivity, media cleanup) are lower priority and can come in follow-ups.

I am also watching #619 and #639 which both feed into this refactor. The FileMediaStore with scope-based lifecycle is the right abstraction — just needs the crash-recovery safety net (TTL scanner) that @ex-takashima proposed in #620.

Happy to review the next iteration.