​
Digital Worker Backend Implementation

​
Overview

Digital Worker (“Team Chat Observer”) is a chat-resident AI agent. It watches team chat channels (Telegram, WhatsApp, Discord), decides — cheaply and locally first — whether a message is worth acting on, and only then spends a billed LLM call to draft a follow-up into a human-reviewed worker inbox. Approved items can be promoted into Fieldforce tasks. Everything is behind the two-tier digital_worker feature flag (seeded OFF), so the whole module dark-launches safely.
This page is the module map — use it to see what has been built without reading every file. Each phase below was shipped as its own migration + wiring block in module.go.

​
Feature phases (what’s built)

​
Architecture

Hexagonal / clean architecture (Ports & Adapters + DDD), the same shape as leave and fieldforce. Root: internal/modules/digitalworker/.
  • domain/entity/ — ChatEvent/ChatEventRecord, ExternalIdentity, ChannelBinding, RoleAgent, InboxItem, AgentSoul, memory/skill/learning entities, Digest, VerificationCode, MessageEmbedding/PoisonWindow.
  • application/port/ — repository + service interfaces (ChatEventRepository, FeatureFlag, ConfigService, Embedder, MessageEmbeddingRepository, …).
  • application/usecase/ — business logic (ingest, pipeline, policy, inbox, learning, digest, retention, embedding sweep, hybrid search).
  • adapter/inbound/ — http/ (Echo handlers), gateway/ (Discord WebSocket), normalize/ (provider payload → ChatEvent).
  • adapter/outbound/ — ai/ (local Ollama triage + embedder, DB-managed prompts), persistence/postgresql/, external/ (chat senders), storage/ (raw-payload object store), acl/ (Fieldforce promotion, notifier).
  • module.go — wires all of the above; exposes Retention() and EmbeddingSweep() for the platform scheduler.

​
Core ingest → action flow

  1. Provider ingress — Telegram/WhatsApp webhooks (public, signature-checked) or the Discord Gateway socket (single-owner advisory lease, ADR-0031) → normalize.* → ChatEvent.
  2. IngestService.Ingest — kill-switch flag check, idempotency (channel-scoped unique key), optional raw-payload storage behind a reference, edit/delete flagging. Records a dw_chat_events row for every outcome (including default-deny, D6).
  3. PolicyService — default-deny gate: only known identities + bound channels proceed (dw_external_identities, dw_channel_bindings).
  4. Two-tier AI pipeline (Pipeline / EventProcessor):
    • Local triage (OllamaTriage, ADR-0028) — an unbilled local model pre-filter that never routes through AccessGate.
    • Billed remote — only if triage escalates: AccessGate + AuthzRunner + DB-managed prompt (ADR-0025) produce a draft.
  5. Worker inbox (dw_inbox_items) — drafts land for human review; approvers are notified.
  6. Promotion — approved items become Fieldforce tasks via the ACL (TaskPromoter, ADR-0009); the module never writes ff_* directly.
Local triage is the cost guard: most chatter is filtered for free, and only messages with a probable commitment ever cost a billed call.

​
Data model (dw_* tables)

  • Agents & workers — dw_role_agents, dw_workers, dw_worker_credentials (encrypted tokens), dw_channel_bindings, dw_gateway_sessions (Discord resume state).
  • Identity — dw_external_identities, dw_verification_codes (Phase 1.5).
  • Ingest log — dw_chat_events (with a search_tsv FTS column), dw_ai_runs.
  • Inbox — dw_inbox_items, dw_inbox_item_sources.
  • Learning (Phase 2) — dw_agent_souls, dw_agent_memories, dw_agent_skills, dw_agent_skill_versions, dw_agent_managers, dw_learning_items, dw_learning_events.
  • Digest (Phase 3) — dw_digests.
  • Semantic search — dw_message_embeddings (halfvec(1024)), dw_embedding_watermarks, dw_embedding_poison_windows.
  • Shared — per-org config in org_module_configs (module = digital_worker); audit in the shared audit_log.

​
Feature flags (ADR-0016)

Two-tier: global admin_feature_flags (kill switch) AND per-org org_feature_flags. Both must be on; both are seeded OFF.
  • digital_worker — the whole module (ingest, pipeline, routes, Discord reconcile).
  • digital_worker_semantic_search — hybrid semantic history search, checked in addition to digital_worker.

​
Semantic history search (ADR-0050)

The simple FTS config does no Chinese segmentation and no stemming, so DW’s code-switched (“rojak”) Malay/Chinese chat is effectively unsearchable. Local embeddings on the existing triage Ollama host fix this at no API cost.
  • Write path — EmbeddingSweep is a periodic watermark sweep: it sessionizes closed conversation windows from dw_chat_events (30-min gap, 15-message cap), applies a window-level substance filter (~20 meaningful chars), embeds each closed window with the local model, and advances a per-(org, provider, channel, model) watermark. The watermark alone gives async embedding, retry (stall on transient failure), backfill (empty watermark), and model-swap rebuild. Deterministic failures are recorded as poison windows and skipped. Ingest is never touched.
  • Read path — HistoryService.SearchAgentHistory is hybrid: FTS first, semantic KNN appended, deduped by event ID, capped at the limit, and it degrades to FTS-only on any embedding failure. Every KNN query filters WHERE org_id = ? AND model = ? (model identity defines the vector space). No new endpoint.
  • Storage — dw_message_embeddings with halfvec(1024) + exact cosine KNN (ORDER BY embedding <=> ?). No ANN index — retention bounds the table to hundreds–low-thousands of vectors/org where exact scan is single-digit ms.
  • Config — deployment env DIGITAL_WORKER_EMBED_MODEL (default bge-m3), reusing the triage OllamaURL. Not per-org.
Requires pgvector ≥ 0.7 (halfvec). Rollout gate per environment: provision the embed model on Ollama, pass the cross-lingual smoke test, then flip digital_worker_semantic_search on.

​
Retention & privacy

RetentionSweep (driven by the platform scheduler) enforces per-org windows from org_module_configs:
  • Raw payloads deleted from object storage past raw_retention_days (ref cleared).
  • dw_chat_events purged past event_retention_days.
  • dw_message_embeddings purged at the same cutoff as their source events — embedded chunk text must not outlive event retention, so semantic search depth equals the retention window.

​
Future phases

Planned work, not yet built — recorded here so the roadmap stays visible.

​
Media semantic search (ADR-0051)

Today, semantic search is text-only. A WhatsApp image/voice/video is normalized to a placeholder ([image], [audio]) — the actual content is never embedded, and media-only windows are skipped by the substance filter. Captions are not even captured yet. The planned approach (Strategy A: media → text → existing text embeddings, ADR-0051) makes media searchable without adding a second vector space:
  • Voice/audio → local ASR (Whisper) transcript.
  • Image → local vision model (Ollama llama3.2-vision / qwen2.5-vl): OCR the text and describe the picture.
  • Video → keyframes (vision) + audio track (ASR).
  • Documents/PDF → text extraction (OCR fallback).
Enrichment runs off the ingest hot path as an async watermark sweep (same shape as EmbeddingSweep), writes a derived_text column on dw_chat_events (original placeholder + raw-payload reference kept for provenance), and the existing embedding sweep then embeds that text — so dw_message_embeddings, the KNN, the hybrid read path, and retention are unchanged. Delivery is phased: voice notes → images → video/documents. See ADR-0051 for the full decision and rejected alternatives (native multimodal embeddings, paid multimodal APIs, inline enrichment).

​
Also deferred

  • Local reranker (bge-reranker-v2-m3) as a separate inference service to sharpen semantic ranking precision (noted in ADR-0050 as phase 2).

​
Key files

  • Wiring: internal/modules/digitalworker/module.go
  • Ingest / policy: application/usecase/{ingest_service,policy_service}.go
  • AI pipeline: application/usecase/{pipeline,event_processor}.go; adapter/outbound/ai/{local_triage,local_embedder,prompt_provider}.go
  • Inbox / promotion: application/usecase/inbox_service.go; adapter/outbound/acl/
  • Learning: application/usecase/{manage_agent_soul,save_memory,skill_service,learning_inbox_actions,detect_learning}.go
  • Digest: application/usecase/{digest_scheduler,generate_channel_digest}.go
  • Semantic search: application/usecase/{embedding_window,embedding_sweep,search_history,retention_sweep}.go; adapter/outbound/persistence/postgresql/embedding_repository.go
  • Migrations: internal/shared/infrastructure/database/postgresql/migrations/migrator.go (CreateDigitalWorker*Tables)
  • 0016 two-tier feature flags · 0025 DB-managed prompts · 0028 local triage cost guard · 0031 single-owner gateway lease · 0034 defer-AI / recency-ranked memory · 0035/0036 self-service verification · 0050 local embeddings for semantic history search · 0051 media semantic search via text conversion (planned)