"EMAIL DSAR message export — every message a subject authored or received + chained memory audit hashes for TASK-PORTAL-008 bundle"

§1 — Description (BCP-14 normative)

The EMAIL service MUST ship DSAR message export at services/email/src/dsar/ returning subject-scoped JSONL of every authored + received + cc'd message with attachment S3 refs + per-message memory chain anchor, async via TASK-MCP-007, 3 memory audit kinds.

1. MUST expose POST /v1/email/dsar/export body { subject_id }. Caller is TASK-PORTAL-008 task (system-tenant). Enqueues TASK-MCP-007 task per DEC-1504; returns task_id.

1. MUST aggregate via aggregator.rs::aggregate(tenant_id, subject_id):

• SELECT messages WHERE author_subject_id=$subject AND tenant_id=$tenant.
• UNION SELECT WHERE recipient or cc or bcc matches subject's email_addresses.
• Per DEC-1503: cross-tenant NEVER included.

1. MUST write JSONL per DEC-1501 — one message per line, structure: { id, from, to, cc, subject, body_text, body_html, sent_at, attachments: [{filename, s3_key, sha256, size}], memory_audit_chain_hash }.

1. MUST include attachment S3 references per DEC-1501 — never inline bytes (size + bundle bloat).

1. MUST include memory chain anchor per DEC-1502 via chain_anchor.rs::lookup(message_id) — finds the email.send_queued or email.message_received memory row for the message; embeds chain hash.

1. MUST define dsar_export_jobs table at migration 0007: (job_id UUID PRIMARY KEY, tenant_id UUID NOT NULL, subject_id UUID NOT NULL, status TEXT NOT NULL DEFAULT 'pending', message_count INT, attachment_count INT, output_s3_key TEXT, started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, trace_id CHAR(32)). Append-only.

1. MUST stream output to S3 (TASK-DOC-001 path); never load entire mailbox into memory.

1. MUST emit 3 memory audit kinds per DEC-1505.

1. MUST thread trace_id from TASK-PORTAL-008 task through aggregator + writer.

1. MUST NOT include cross-tenant per DEC-1503.

1. MUST NOT inline attachment bytes per DEC-1501.

§2 — Why this design

Why JSONL + S3 refs (DEC-1501)? Streaming-friendly format; large mailboxes don't OOM. S3 refs preserve attachments without bundle bloat.

Why chain anchor per message (DEC-1502)? Tamper-evident: recipient can prove message wasn't altered post-export.

Why async (DEC-1504)? Large mailbox aggregation is minutes; sync would timeout.

§3 — API contract

CREATE TABLE dsar_export_jobs (
  job_id UUID PRIMARY KEY,
  tenant_id UUID NOT NULL,
  subject_id UUID NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','running','completed','failed')),
  message_count INT,
  attachment_count INT,
  output_s3_key TEXT,
  started_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  failure_reason TEXT,
  trace_id CHAR(32)
);
ALTER TABLE dsar_export_jobs ENABLE ROW LEVEL SECURITY;
CREATE POLICY dsar_jobs_rls ON dsar_export_jobs
  USING (tenant_id = current_setting('auth.tenant_id')::uuid)
  WITH CHECK (tenant_id = current_setting('auth.tenant_id')::uuid);
REVOKE UPDATE, DELETE ON dsar_export_jobs FROM cyberos_app;
GRANT UPDATE (status, message_count, attachment_count, output_s3_key,
              started_at, completed_at, failure_reason) ON dsar_export_jobs TO cyberos_app;

Endpoints:

POST   /v1/email/dsar/export          (system-tenant via TASK-PORTAL-008)
GET    /v1/email/dsar/jobs/{id}       (poll status)

§4 — Acceptance criteria

1. All authored messages included. 2. All received messages included. 3. CC/BCC matches included. 4. Cross-tenant excluded. 5. Attachments by S3 ref only. 6. Chain anchor present per message. 7. JSONL format valid. 8. Async via TASK-MCP-007. 9. Status poll returns progress. 10. 3 memory audit kinds emitted. 11. Trace_id from TASK-PORTAL-008 preserved. 12. Large mailbox (10k msgs) completes < 30min. 13. PII not in audit chain (only message counts). 14. RLS denies non-system caller. 15. Job idempotency — duplicate request returns existing job_id. 16. Stream to S3 (no OOM). 17. Chain anchor missing → noted in output. 18. Per-message size cap respected. 19. Output S3 key persistent. 20. Failure path emits sev-2.

§5 — Verification

#[tokio::test]
async fn dsar_exports_authored_and_received() {
    let ctx = TestContext::with_subject_messages(100, 50).await;  // 100 authored, 50 received
    let job = ctx.start_dsar_export(ctx.subject_id).await;
    ctx.wait_completion(job).await;
    let s3_key: String = ctx.get_job_output(job).await;
    let lines = ctx.read_s3_jsonl(&s3_key).await;
    assert_eq!(lines.len(), 150);
}

#[tokio::test]
async fn cross_tenant_excluded() {
    let ctx = TestContext::with_subject_in_two_tenants().await;
    ctx.send_message_in_tenant(ctx.tenant_a, ctx.subject_id, "in-a").await;
    ctx.send_message_in_tenant(ctx.tenant_b, ctx.subject_id, "in-b").await;
    let job = ctx.start_dsar_export_for(ctx.tenant_a, ctx.subject_id).await;
    let lines = ctx.read_s3_jsonl_from_job(job).await;
    assert!(lines.iter().any(|m| m.contains("in-a")));
    assert!(!lines.iter().any(|m| m.contains("in-b")));
}

#[tokio::test]
async fn chain_anchor_per_message() {
    let ctx = TestContext::with_subject_messages(5, 0).await;
    let job = ctx.complete_dsar().await;
    let lines = ctx.read_s3_jsonl_from_job(job).await;
    for line in lines {
        let msg: serde_json::Value = serde_json::from_str(&line).unwrap();
        assert!(msg["memory_audit_chain_hash"].is_string());
    }
}

// 5.4..5.10

§7 — Dependencies

Upstream: TASK-EMAIL-001. Cross-module: TASK-PORTAL-008 (caller), TASK-MCP-007 (async task), TASK-DOC-001 (S3), TASK-AI-003, TASK-MEMORY-111.

§10 — Failure modes

§11 — Implementation notes

• §11.1 JSONL streaming via tokio AsyncWriter to S3 multipart upload.
• §11.2 Chain anchor lookup batched (in chunks of 1000 messages) for performance.
• §11.3 Aggregator uses SELECT with UNION + tenant + subject filters; indexed by (tenant_id, author_subject_id) and (tenant_id, recipient_subject_id).
• §11.4 Per-message size cap 25 MiB (consistent with attachment cap).
• §11.5 PII: message bodies not in memory chain; only counts.

End of TASK-EMAIL-011 spec.