Compare commits

...

5 Commits

Author SHA1 Message Date
fullstack-engineer 1b9e69b309 test(handlers): add queueRowAuthFields + additional GetA2AQueueStatus coverage
CI / all-required (pull_request) Blocked by required conditions
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 31s
CI / Detect changes (pull_request) Successful in 34s
Harness Replays / detect-changes (pull_request) Successful in 32s
E2E API Smoke Test / detect-changes (pull_request) Successful in 46s
Handlers Postgres Integration / detect-changes (pull_request) Successful in 46s
Secret scan / Scan diff for credential-shaped strings (pull_request) Successful in 37s
Runtime PR-Built Compatibility / detect-changes (pull_request) Successful in 51s
gate-check-v3 / gate-check (pull_request) Successful in 33s
qa-review / approved (pull_request) Successful in 35s
sop-tier-check / tier-check (pull_request) Successful in 36s
sop-checklist / all-items-acked (pull_request) Successful in 40s
lint-required-no-paths / lint-required-no-paths (pull_request) Successful in 1m47s
CI / Canvas (Next.js) (pull_request) Failing after 10m22s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
CI / Platform (Go) (pull_request) Failing after 10m44s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 12s
CI / Python Lint & Test (pull_request) Successful in 11s
Harness Replays / Harness Replays (pull_request) Successful in 7s
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Successful in 13s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Failing after 6m55s
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Failing after 7m4s
security-review / approved (pull_request) Has been cancelled
Extends fix/a2a-queue-status-coverage with:
- TestQueueRowAuthFields_Success_BothPresent (internal helper success path)
- TestQueueRowAuthFields_NoRows_ReturnsErrNoRows
- TestQueueRowAuthFields_QueryError_ReturnsError
- TestGetA2AQueueStatus_AuthPass_CallerMatchesCallerID (caller_id auth path)
- TestGetA2AQueueStatus_AuthPass_OrgTokenBypassesAuth (org-level token bypass)
- TestGetA2AQueueStatus_StatusQueryNoRows_NotFound (race-to-404)
- TestGetA2AQueueStatus_ResponseBodyIncludedWhenCompleted

All 30 platform packages pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-16 14:23:21 +00:00
fullstack-engineer 0967ed908d test(handlers): add coverage for QueueDepth, QueueStatusByID, GetA2AQueueStatus, emitOrgEvent nil-payload
a2a_queue_status.go had 0% coverage across all 3 exported symbols.
Added 14 tests exercising:

QueueDepth (package-level):
- TestQueueDepth_Success: COUNT returns 7
- TestQueueDepth_EmptyQueue: COUNT returns 0
- TestQueueDepth_QueryError_ReturnsZero: DB error → returns 0 (non-fatal)

QueueStatusByID (package-level):
- TestQueueStatusByID_Success: fully-populated QueueStatus from LEFT JOIN
- TestQueueStatusByID_CompletedWithResponse: completed item populates ResponseBody
- TestQueueStatusByID_ErrNoRows: sql.ErrNoRows propagates
- TestQueueStatusByID_QueryError: DB error propagates

GetA2AQueueStatus (HTTP handler):
- TestGetA2AQueueStatus_MissingQueueID_Returns400
- TestGetA2AQueueStatus_NoIdentity_Returns404 (not 401 per design)
- TestGetA2AQueueStatus_QueueNotFound_Returns404
- TestGetA2AQueueStatus_UnauthorizedCaller_Returns404 (not 403 per design)
- TestGetA2AQueueStatus_AuthorizedAsTarget_Success
- TestGetA2AQueueStatus_QueueRowLookupError_Returns500
- TestGetA2AQueueStatus_StatusFetchError_Returns500

org_import_reconcile_test.go:
- TestEmitOrgEvent_NilPayloadInitializesEmptyMap: exercises the
  payload == nil branch so the empty-map init path is covered.

All tests pass; full suite: 69.1% → 69.7%.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-16 14:23:21 +00:00
devops-engineer b5411d2c37 Merge pull request 'harden(provisioner): denylist SCM-write tokens from tenant workspace env (forensic #145)' (#1277) from harden/provisioner-scm-token-denylist into staging
Block internal-flavored paths / Block forbidden paths (push) Successful in 17s
Harness Replays / detect-changes (push) Successful in 19s
CI / Detect changes (push) Successful in 1m14s
E2E Chat / detect-changes (push) Successful in 1m25s
E2E API Smoke Test / detect-changes (push) Successful in 1m28s
Secret scan / Scan diff for credential-shaped strings (push) Successful in 26s
Handlers Postgres Integration / detect-changes (push) Successful in 1m18s
Runtime PR-Built Compatibility / detect-changes (push) Successful in 1m9s
CI / Shellcheck (E2E scripts) (push) Successful in 8s
Harness Replays / Harness Replays (push) Successful in 8s
CI / Python Lint & Test (push) Successful in 10s
E2E Chat / E2E Chat (push) Failing after 44s
Runtime PR-Built Compatibility / PR-built wheel + import smoke (push) Failing after 1m15s
E2E API Smoke Test / E2E API Smoke Test (push) Successful in 2m17s
Handlers Postgres Integration / Handlers Postgres Integration (push) Successful in 5m57s
CI / Canvas (Next.js) (push) Successful in 18m50s
CI / Platform (Go) (push) Successful in 19m21s
CI / Canvas Deploy Reminder (push) Successful in 6s
CI / all-required (push) Has been cancelled
2026-05-16 05:16:21 +00:00
claude-ceo-assistant 03ad7ab2d8 chore(ci): re-trigger required checks (post-#441 fix; 03:50Z storm-cancel residue)
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 18s
CI / Detect changes (pull_request) Successful in 1m9s
Harness Replays / detect-changes (pull_request) Successful in 29s
Secret scan / Scan diff for credential-shaped strings (pull_request) Successful in 29s
E2E API Smoke Test / detect-changes (pull_request) Successful in 1m26s
E2E Chat / detect-changes (pull_request) Successful in 1m23s
gate-check-v3 / gate-check (pull_request) Successful in 25s
Handlers Postgres Integration / detect-changes (pull_request) Successful in 1m29s
qa-review / approved (pull_request) Successful in 29s
security-review / approved (pull_request) Successful in 29s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 8s
sop-checklist / all-items-acked (pull_request) Successful in 33s
sop-tier-check / tier-check (pull_request) Successful in 26s
CI / Python Lint & Test (pull_request) Successful in 12s
Harness Replays / Harness Replays (pull_request) Successful in 11s
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Successful in 20s
lint-required-no-paths / lint-required-no-paths (pull_request) Successful in 1m44s
E2E Chat / E2E Chat (pull_request) Failing after 23s
Runtime PR-Built Compatibility / detect-changes (pull_request) Successful in 1m46s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Successful in 2m22s
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Successful in 7s
CI / Canvas (Next.js) (pull_request) Successful in 18m41s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
CI / Platform (Go) (pull_request) Successful in 19m37s
CI / all-required (pull_request) Successful in 5s
audit-force-merge / audit (pull_request) Successful in 25s
2026-05-15 21:40:47 -07:00
core-be fd545a332b harden(provisioner): denylist SCM-write tokens from tenant workspace env (forensic #145)
E2E API Smoke Test / detect-changes (pull_request) Waiting to run
E2E Chat / detect-changes (pull_request) Waiting to run
Handlers Postgres Integration / detect-changes (pull_request) Waiting to run
Harness Replays / detect-changes (pull_request) Waiting to run
lint-required-no-paths / lint-required-no-paths (pull_request) Waiting to run
Runtime PR-Built Compatibility / detect-changes (pull_request) Waiting to run
Secret scan / Scan diff for credential-shaped strings (pull_request) Waiting to run
gate-check-v3 / gate-check (pull_request) Waiting to run
qa-review / approved (pull_request) Waiting to run
security-review / approved (pull_request) Waiting to run
sop-checklist / all-items-acked (pull_request) Waiting to run
sop-tier-check / tier-check (pull_request) Waiting to run
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 28s
CI / Detect changes (pull_request) Successful in 3m54s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 6s
CI / Python Lint & Test (pull_request) Successful in 8s
CI / Canvas (Next.js) (pull_request) Successful in 24m25s
CI / Platform (Go) (pull_request) Successful in 26m22s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
CI / all-required (pull_request) Successful in 6s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Has been cancelled
E2E Chat / E2E Chat (pull_request) Has been cancelled
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Has been cancelled
Harness Replays / Harness Replays (pull_request) Has been cancelled
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Has been cancelled
Tenant workspace containers run agent-controlled code and must never
receive a Git SCM write credential — agents structurally lacking
merge/approve creds is why the two-eyes review gate is self-bypass-proof
against forged-approval injection.

Latent path: handlers.loadPersonaEnvFile() merges a per-role persona
GITEA_TOKEN into cfg.EnvVars when MOLECULE_PERSONA_ROOT is set on a
tenant host; it then flowed unfiltered through buildContainerEnv()
(local Docker) and CPProvisioner.Start() (tenant EC2). Inert today
(persona dirs are operator-host-only) but unguarded — and the
pre-existing TestBuildContainerEnv_CustomEnvVarsAppended test actually
asserted GITHUB_TOKEN passed through verbatim.

Adds a narrow, auditable exact-match denylist (isSCMWriteTokenKey:
GITEA/GITHUB/GH/GITLAB/GL/BITBUCKET _TOKEN) applied by construction in
both env paths, plus negative-assertion tests covering the normal path
and a persona-file-merge simulation. Non-credential persona identity
(GITEA_USER, GITEA_USER_EMAIL) is intentionally preserved. No
provisioner refactor.

Tracking: molecule-ai/internal#438

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-15 18:43:18 -07:00
5 changed files with 806 additions and 9 deletions
@@ -1,7 +1,16 @@
package handlers
import (
"context"
"database/sql"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"testing"
"github.com/DATA-DOG/go-sqlmock"
"github.com/gin-gonic/gin"
)
// TestExtractExpiresInSeconds covers the JSON parser used at enqueue time
@@ -58,3 +67,597 @@ func TestExtractExpiresInSeconds(t *testing.T) {
})
}
}
// ─── QueueDepth ─────────────────────────────────────────────────────────────
// TestQueueDepth_Success verifies QueueDepth returns the COUNT of queued items
// for a workspace.
func TestQueueDepth_Success(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM a2a_queue WHERE workspace_id = \$1 AND status = 'queued'`).
WithArgs("ws-queue-depth-1").
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(7))
got := QueueDepth(context.Background(), "ws-queue-depth-1")
if got != 7 {
t.Errorf("QueueDepth() = %d; want 7", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestQueueDepth_EmptyQueue returns 0 when no queued items exist.
func TestQueueDepth_EmptyQueue(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM a2a_queue WHERE workspace_id = \$1 AND status = 'queued'`).
WithArgs("ws-empty").
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
got := QueueDepth(context.Background(), "ws-empty")
if got != 0 {
t.Errorf("QueueDepth() = %d; want 0", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestQueueDepth_QueryError returns 0 on DB error (non-fatal; caller only uses
// the count for display purposes).
func TestQueueDepth_QueryError_ReturnsZero(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM a2a_queue WHERE workspace_id = \$1 AND status = 'queued'`).
WithArgs("ws-err").
WillReturnError(errors.New("connection refused"))
// QueueDepth swallows the error and returns 0.
got := QueueDepth(context.Background(), "ws-err")
if got != 0 {
t.Errorf("QueueDepth() on error = %d; want 0", got)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// ─── QueueStatusByID ────────────────────────────────────────────────────────
// TestQueueStatusByID_Success verifies QueueStatusByID returns a fully-populated
// QueueStatus from the LEFT JOIN of a2a_queue and activity_logs.
func TestQueueStatusByID_Success(t *testing.T) {
mock := setupTestDB(t)
// The LEFT JOIN query returns all queue columns + NULL for activity_logs
// when no delegation row exists.
mock.ExpectQuery(`SELECT\s+q\.id,\s+q\.workspace_id,\s+q\.status,\s+q\.priority,\s+q\.attempts,\s+q\.last_error,\s+q\.enqueued_at::text,\s+q\.dispatched_at::text,\s+q\.completed_at::text,\s+q\.expires_at::text,\s+al\.response_body::text\s+FROM a2a_queue q\s+LEFT JOIN activity_logs al`).
WithArgs("queue-ok-1").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"queue-ok-1", "ws-1", "queued", 50, 1,
nil, "2026-05-16T10:00:00Z", nil, nil, "2026-05-16T12:00:00Z",
nil,
))
qs, err := QueueStatusByID(context.Background(), "queue-ok-1")
if err != nil {
t.Fatalf("QueueStatusByID() error = %v; want nil", err)
}
if qs.ID != "queue-ok-1" {
t.Errorf("ID = %q; want queue-ok-1", qs.ID)
}
if qs.WorkspaceID != "ws-1" {
t.Errorf("WorkspaceID = %q; want ws-1", qs.WorkspaceID)
}
if qs.Status != "queued" {
t.Errorf("Status = %q; want queued", qs.Status)
}
if qs.Priority != 50 {
t.Errorf("Priority = %d; want 50", qs.Priority)
}
if qs.Attempts != 1 {
t.Errorf("Attempts = %d; want 1", qs.Attempts)
}
if qs.LastError != nil {
t.Errorf("LastError = %v; want nil", qs.LastError)
}
if qs.EnqueuedAt != "2026-05-16T10:00:00Z" {
t.Errorf("EnqueuedAt = %q; want 2026-05-16T10:00:00Z", qs.EnqueuedAt)
}
if qs.DispatchedAt != nil {
t.Errorf("DispatchedAt = %v; want nil", qs.DispatchedAt)
}
if qs.CompletedAt != nil {
t.Errorf("CompletedAt = %v; want nil", qs.CompletedAt)
}
if *qs.ExpiresAt != "2026-05-16T12:00:00Z" {
t.Errorf("ExpiresAt = %v; want 2026-05-16T12:00:00Z", qs.ExpiresAt)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestQueueStatusByID_CompletedWithResponse verifies that a completed queue item
// populates ResponseBody from the LEFT JOINed activity_logs row.
func TestQueueStatusByID_CompletedWithResponse(t *testing.T) {
mock := setupTestDB(t)
respBody := `{"result":"done"}`
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("queue-done-1").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"queue-done-1", "ws-1", "completed", 50, 1,
nil, "2026-05-16T10:00:00Z", "2026-05-16T10:01:00Z", "2026-05-16T10:02:00Z", nil,
respBody,
))
qs, err := QueueStatusByID(context.Background(), "queue-done-1")
if err != nil {
t.Fatalf("QueueStatusByID() error = %v; want nil", err)
}
if qs.Status != "completed" {
t.Errorf("Status = %q; want completed", qs.Status)
}
if qs.ResponseBody == nil {
t.Fatal("ResponseBody = nil; want non-nil for completed item")
}
var resp map[string]interface{}
if err := json.Unmarshal(qs.ResponseBody, &resp); err != nil {
t.Fatalf("ResponseBody not valid JSON: %v", err)
}
if resp["result"] != "done" {
t.Errorf("ResponseBody result = %v; want done", resp["result"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestQueueStatusByID_ErrNoRows returns sql.ErrNoRows when the queue ID doesn't exist.
func TestQueueStatusByID_ErrNoRows(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("queue-missing").
WillReturnError(sql.ErrNoRows)
_, err := QueueStatusByID(context.Background(), "queue-missing")
if !errors.Is(err, sql.ErrNoRows) {
t.Errorf("QueueStatusByID() error = %v; want sql.ErrNoRows", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestQueueStatusByID_QueryError propagates DB errors as-is.
func TestQueueStatusByID_QueryError(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("queue-err").
WillReturnError(errors.New("connection refused"))
_, err := QueueStatusByID(context.Background(), "queue-err")
if err == nil {
t.Fatal("QueueStatusByID() error = nil; want non-nil")
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// ─── GetA2AQueueStatus (HTTP handler) ─────────────────────────────────────
func newGetA2AQueueStatusHarness(t *testing.T) (sqlmock.Sqlmock, *httptest.ResponseRecorder, *gin.Context) {
mock := setupTestDB(t)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
return mock, w, c
}
func TestGetA2AQueueStatus_MissingQueueID_Returns400(t *testing.T) {
_, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: ""}}
c.Request = httptest.NewRequest("GET", "/", nil)
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusBadRequest {
t.Errorf("expected 400, got %d: %s", w.Code, w.Body.String())
}
}
func TestGetA2AQueueStatus_NoIdentity_Returns404(t *testing.T) {
_, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-123"}}
c.Request = httptest.NewRequest("GET", "/", nil)
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
// Returns 404 (not 401) per the existence-non-inference policy.
if w.Code != http.StatusNotFound {
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
}
}
func TestGetA2AQueueStatus_QueueNotFound_Returns404(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-404"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-1")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-404").
WillReturnError(sql.ErrNoRows)
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusNotFound {
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
func TestGetA2AQueueStatus_UnauthorizedCaller_Returns404(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-unauth"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-wrong")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-unauth").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller-a", "ws-target-b"))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
// Returns 404 per the existence-non-inference policy.
if w.Code != http.StatusNotFound {
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
func TestGetA2AQueueStatus_AuthorizedAsTarget_Success(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-ok"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-target")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-ok").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller", "ws-target"))
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-ok").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"q-ok", "ws-target", "queued", 50, 1,
nil, "2026-05-16T10:00:00Z", nil, nil, nil,
nil,
))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var qs QueueStatus
if err := json.Unmarshal(w.Body.Bytes(), &qs); err != nil {
t.Fatalf("body parse: %v", err)
}
if qs.ID != "q-ok" {
t.Errorf("queue_id = %q; want q-ok", qs.ID)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
func TestGetA2AQueueStatus_QueueRowLookupError_Returns500(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-lookup-err"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-1")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-lookup-err").
WillReturnError(errors.New("connection refused"))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusInternalServerError {
t.Errorf("expected 500, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
func TestGetA2AQueueStatus_StatusFetchError_Returns500(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-status-err"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-1")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-status-err").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-1", "ws-1"))
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-status-err").
WillReturnError(errors.New("connection refused"))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusInternalServerError {
t.Errorf("expected 500, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// ─── queueRowAuthFields (internal helper) ─────────────────────────────────────
// Covers the auth-only 2-col SELECT used by GetA2AQueueStatus to determine
// whether the caller has access before projecting the public status fields.
func TestQueueRowAuthFields_Success_BothPresent(t *testing.T) {
mock := setupTestDB(t)
queueID := "qqqqqqqq-0003-0003-0003-000000000003"
rows := sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller-3", "ws-target-3")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs(queueID).
WillReturnRows(rows)
callerID, workspaceID, err := queueRowAuthFields(context.Background(), queueID)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if callerID != "ws-caller-3" {
t.Errorf("callerID = %q, want %q", callerID, "ws-caller-3")
}
if workspaceID != "ws-target-3" {
t.Errorf("workspaceID = %q, want %q", workspaceID, "ws-target-3")
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueRowAuthFields_NoRows_ReturnsErrNoRows(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("qqqqqqqq-missing").
WillReturnError(sql.ErrNoRows)
_, _, err := queueRowAuthFields(context.Background(), "qqqqqqqq-missing")
if !errors.Is(err, sql.ErrNoRows) {
t.Errorf("expected sql.ErrNoRows, got %v", err)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
func TestQueueRowAuthFields_QueryError_ReturnsError(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("qqqqqqqq-dberr").
WillReturnError(sql.ErrConnDone)
_, _, err := queueRowAuthFields(context.Background(), "qqqqqqqq-dberr")
if err == nil {
t.Fatal("expected error, got nil")
}
if errors.Is(err, sql.ErrNoRows) {
t.Error("expected non-no-rows error, got sql.ErrNoRows")
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
}
// ─── Additional GetA2AQueueStatus coverage ─────────────────────────────────────
// TestGetA2AQueueStatus_AuthPass_CallerMatchesCallerID verifies that a caller
// whose workspace matches queue.caller_id (not just workspace_id) passes auth
// and receives the status. This path is distinct from the existing "authorized
// as target" test which covers workspace_id = caller.
func TestGetA2AQueueStatus_AuthPass_CallerMatchesCallerID(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-caller-match"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-caller-match")
// Queue row: ws-caller-match is the caller, ws-other-target is the target.
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-caller-match").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller-match", "ws-other-target"))
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-caller-match").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"q-caller-match", "ws-other-target", "queued", 50, 0,
nil, "2026-05-16T10:00:00Z", nil, nil, nil,
nil,
))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var qs QueueStatus
json.Unmarshal(w.Body.Bytes(), &qs)
if qs.ID != "q-caller-match" {
t.Errorf("queue_id = %q; want q-caller-match", qs.ID)
}
if qs.Status != "queued" {
t.Errorf("status = %q; want queued", qs.Status)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestGetA2AQueueStatus_AuthPass_OrgTokenBypassesAuth verifies that an org-level
// token (canvas/admin) bypasses the caller_id / workspace_id match entirely.
// No X-Workspace-ID header is required; org_token_id in context is sufficient.
func TestGetA2AQueueStatus_AuthPass_OrgTokenBypassesAuth(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-org-bypass"}}
c.Request = httptest.NewRequest("GET", "/", nil)
// No X-Workspace-ID header — org token is set via context instead.
c.Set("org_token_id", "org-admin-1")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-org-bypass").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-anyone", "ws-anyone"))
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-org-bypass").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"q-org-bypass", "ws-anyone", "queued", 25, 0,
nil, "2026-05-16T10:00:00Z", nil, nil, nil,
nil,
))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestGetA2AQueueStatus_StatusQueryNoRows_NotFound covers the theoretical race:
// queue row exists (auth check passes), but is deleted before QueueStatusByID runs.
// Handler returns 404 (not 500) — matching the existence-non-inference policy.
func TestGetA2AQueueStatus_StatusQueryNoRows_NotFound(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-race-no-rows"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-caller")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-race-no-rows").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller", "ws-target"))
// Status query returns no rows — row was deleted between auth check and status fetch.
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-race-no-rows").
WillReturnError(sql.ErrNoRows)
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusNotFound {
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
// TestGetA2AQueueStatus_ResponseBodyIncludedWhenCompleted confirms that a completed
// queue item surfaces response_body from activity_logs in the HTTP response body.
func TestGetA2AQueueStatus_ResponseBodyIncludedWhenCompleted(t *testing.T) {
mock, w, c := newGetA2AQueueStatusHarness(t)
c.Params = gin.Params{{Key: "id", Value: "ws-1"}, {Key: "queue_id", Value: "q-completed-body"}}
c.Request = httptest.NewRequest("GET", "/", nil)
c.Request.Header.Set("X-Workspace-ID", "ws-caller")
mock.ExpectQuery(`SELECT caller_id, workspace_id FROM a2a_queue WHERE id = \$1`).
WithArgs("q-completed-body").
WillReturnRows(sqlmock.NewRows([]string{"caller_id", "workspace_id"}).
AddRow("ws-caller", "ws-target"))
respBody := `{"result":{"status":"ok","reply":"hello world"}}`
mock.ExpectQuery(`SELECT\s+q\.id`).
WithArgs("q-completed-body").
WillReturnRows(sqlmock.NewRows([]string{
"id", "workspace_id", "status", "priority", "attempts",
"last_error", "enqueued_at", "dispatched_at", "completed_at", "expires_at",
"response_body",
}).AddRow(
"q-completed-body", "ws-target", "completed", 50, 1,
nil, "2026-05-16T10:00:00Z", "2026-05-16T10:01:00Z", "2026-05-16T10:02:00Z", nil,
respBody,
))
h := newHandlerWithTestDeps(t)
h.GetA2AQueueStatus(c)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var qs QueueStatus
json.Unmarshal(w.Body.Bytes(), &qs)
if qs.ResponseBody == nil {
t.Fatal("ResponseBody should be set for completed status")
}
if string(qs.ResponseBody) != respBody {
t.Errorf("ResponseBody = %q, want %q", string(qs.ResponseBody), respBody)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet: %v", err)
}
}
@@ -156,3 +156,20 @@ func equalStrings(a, b []string) bool {
}
return true
}
// TestEmitOrgEvent_NilPayload exercises the `if payload == nil` branch that
// re-initializes payload to an empty map before marshaling.
func TestEmitOrgEvent_NilPayloadInitializesEmptyMap(t *testing.T) {
mock := setupTestDB(t)
mock.ExpectExec(`INSERT INTO structure_events`).
WithArgs("org.import.started", sqlmock.AnyArg()).
WillReturnResult(sqlmock.NewResult(1, 1))
// Passing nil triggers: if payload == nil { payload = map[string]any{} }
emitOrgEvent(context.Background(), "org.import.started", nil)
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sqlmock expectations: %v", err)
}
}
@@ -178,12 +178,21 @@ func (p *CPProvisioner) Start(ctx context.Context, cfg WorkspaceConfig) (string,
// /admin/liveness and other admin-gated platform endpoints (core#831).
// p.adminToken is read from os.Getenv("ADMIN_TOKEN") at provisioner creation;
// it is also used for CP→platform HTTP auth but those are separate concerns.
env := cfg.EnvVars
if p.adminToken != "" {
env = make(map[string]string, len(cfg.EnvVars)+1)
for k, v := range cfg.EnvVars {
env[k] = v
//
// Forensic #145 hardening: tenant workspaces run on EC2 via this path, so
// the SCM-write-token denylist (see buildContainerEnv) is enforced here
// too. Always build a filtered copy — never pass cfg.EnvVars through
// verbatim — so a latent persona-merged GITEA_TOKEN can't reach the
// tenant container regardless of whether ADMIN_TOKEN is set.
env := make(map[string]string, len(cfg.EnvVars)+1)
for k, v := range cfg.EnvVars {
if isSCMWriteTokenKey(k) {
log.Printf("CPProvisioner.Start: dropped SCM-write credential %q from tenant workspace env (forensic #145 guard)", k)
continue
}
env[k] = v
}
if p.adminToken != "" {
env["ADMIN_TOKEN"] = p.adminToken
}
// Collect template files and generated configs, with OFFSEC-010 guards:
@@ -343,6 +352,7 @@ func collectCPConfigFiles(cfg WorkspaceConfig) (map[string]string, error) {
}
return files, nil
}
// Stop terminates the workspace's EC2 instance via the control plane.
//
// Looks up the actual EC2 instance_id from the workspaces table before
@@ -497,7 +507,9 @@ func (p *CPProvisioner) IsRunning(ctx context.Context, workspaceID string) (bool
// Don't leak the body — upstream errors may echo headers.
return true, fmt.Errorf("cp provisioner: status: unexpected %d", resp.StatusCode)
}
var result struct{ State string `json:"state"` }
var result struct {
State string `json:"state"`
}
// Cap body read at 64 KiB for parity with Start — a misconfigured
// or compromised CP streaming a huge body could otherwise exhaust
// memory in this hot path (called reactively per-request from
@@ -591,6 +591,28 @@ func ValidateWorkspaceAccess(access, workspacePath string) error {
}
}
// scmWriteTokenKeys is the explicit denylist of environment variable names
// that carry a Git SCM *write* credential (push / merge / approve). These
// must never reach a tenant workspace container — see the forensic #145
// rationale in buildContainerEnv. Kept as an exact-match set rather than a
// substring/prefix heuristic so the guard is auditable and can't silently
// over-strip a legitimately-named var.
var scmWriteTokenKeys = map[string]struct{}{
"GITEA_TOKEN": {},
"GITHUB_TOKEN": {},
"GH_TOKEN": {}, // gh CLI honours GH_TOKEN as a GITHUB_TOKEN alias
"GITLAB_TOKEN": {},
"GL_TOKEN": {}, // glab CLI alias
"BITBUCKET_TOKEN": {},
}
// isSCMWriteTokenKey reports whether an env var name is a known Git SCM
// write credential that must be stripped from tenant workspace env.
func isSCMWriteTokenKey(key string) bool {
_, ok := scmWriteTokenKeys[key]
return ok
}
// buildContainerEnv assembles the initial environment variables injected
// into every workspace container.
//
@@ -627,6 +649,21 @@ func buildContainerEnv(cfg WorkspaceConfig) []string {
env = append(env, fmt.Sprintf("AWARENESS_URL=%s", cfg.AwarenessURL))
}
for k, v := range cfg.EnvVars {
// Forensic #145 hardening: tenant workspace containers run
// agent-controlled code and must NEVER receive a Git SCM *write*
// credential. Without merge/approve creds in-container the
// two-eyes review gate is structurally self-bypass-proof — an
// agent that forges an approval has no token to act on it. A
// latent path exists (loadPersonaEnvFile merges a per-role
// persona `GITEA_TOKEN` into cfg.EnvVars when MOLECULE_PERSONA_ROOT
// is set on a tenant host); it is inert today (persona dirs are
// operator-host-only) but unguarded. Strip SCM-write tokens here
// by construction so the invariant holds regardless of whether
// that path ever becomes reachable.
if isSCMWriteTokenKey(k) {
log.Printf("buildContainerEnv: dropped SCM-write credential %q from workspace env (forensic #145 guard)", k)
continue
}
env = append(env, fmt.Sprintf("%s=%s", k, v))
}
// Inject ADMIN_TOKEN from the platform server's environment so workspace
@@ -636,10 +636,15 @@ func TestBuildContainerEnv_AwarenessOnlyWhenBothSet(t *testing.T) {
}
func TestBuildContainerEnv_CustomEnvVarsAppended(t *testing.T) {
// NOTE: this test previously asserted GITHUB_TOKEN passed through
// verbatim. That assertion encoded the forensic #145 latent leak as
// expected behavior. Post-guard, ordinary custom env still flows but
// SCM-write credentials are stripped — see
// TestBuildContainerEnv_StripsSCMWriteTokens for the negative assertion.
cfg := WorkspaceConfig{
WorkspaceID: "ws-x",
PlatformURL: "http://localhost:8080",
EnvVars: map[string]string{"CUSTOM": "value", "GITHUB_TOKEN": "fake-token-for-test"},
EnvVars: map[string]string{"CUSTOM": "value", "ANTHROPIC_API_KEY": "sk-not-an-scm-token"},
}
env := buildContainerEnv(cfg)
seen := map[string]string{}
@@ -652,8 +657,8 @@ func TestBuildContainerEnv_CustomEnvVarsAppended(t *testing.T) {
if seen["CUSTOM"] != "value" {
t.Errorf("CUSTOM env missing, got env=%v", env)
}
if seen["GITHUB_TOKEN"] != "fake-token-for-test" {
t.Errorf("GITHUB_TOKEN env missing, got env=%v", env)
if seen["ANTHROPIC_API_KEY"] != "sk-not-an-scm-token" {
t.Errorf("non-SCM custom env must still pass through, got env=%v", env)
}
// Built-in defaults still present
if seen["MOLECULE_URL"] == "" {
@@ -661,6 +666,129 @@ func TestBuildContainerEnv_CustomEnvVarsAppended(t *testing.T) {
}
}
// ---------- forensic #145: SCM-write-token denylist guard ----------
// TestBuildContainerEnv_StripsSCMWriteTokens is the core negative
// assertion: a tenant workspace env constructed via buildContainerEnv MUST
// NOT contain any Git SCM *write* credential, regardless of how it got into
// cfg.EnvVars. This proves the two-eyes review gate stays structurally
// self-bypass-proof — an agent in-container has no merge/approve token to
// act on a forged approval. See forensic #145.
//
// This test FAILS on the pre-guard code (where buildContainerEnv passed
// cfg.EnvVars through verbatim) and PASSES once the denylist filter is in
// place — i.e. the guard is proven by construction, not by environment
// accident.
func TestBuildContainerEnv_StripsSCMWriteTokens(t *testing.T) {
scmTokens := []string{
"GITEA_TOKEN", "GITHUB_TOKEN", "GH_TOKEN",
"GITLAB_TOKEN", "GL_TOKEN", "BITBUCKET_TOKEN",
}
t.Run("normal path — SCM tokens explicitly set in EnvVars", func(t *testing.T) {
envVars := map[string]string{"CUSTOM": "ok", "ANTHROPIC_API_KEY": "sk-keep"}
for _, k := range scmTokens {
envVars[k] = "leaked-write-credential-" + k
}
cfg := WorkspaceConfig{
WorkspaceID: "ws-tenant",
PlatformURL: "http://localhost:8080",
Tier: 2,
EnvVars: envVars,
}
assertNoSCMWriteToken(t, buildContainerEnv(cfg), scmTokens)
// Sanity: non-SCM custom env is NOT collateral-damaged by the filter.
if !envContains(buildContainerEnv(cfg), "CUSTOM=ok") {
t.Errorf("filter must not strip non-SCM custom env")
}
if !envContains(buildContainerEnv(cfg), "ANTHROPIC_API_KEY=sk-keep") {
t.Errorf("filter must not strip non-SCM API keys")
}
})
t.Run("persona-file path — simulates loadPersonaEnvFile merge", func(t *testing.T) {
// The latent path: handlers.loadPersonaEnvFile() merges a per-role
// persona env file (carrying GITEA_USER, GITEA_TOKEN, …) into the
// workspace env map when MOLECULE_PERSONA_ROOT is set on a tenant
// host. We can't invoke that cross-package helper here, but its
// observable effect is exactly "a GITEA_TOKEN appears in
// cfg.EnvVars". Constructing that condition directly proves the
// guard holds even if the latent path becomes reachable.
cfg := WorkspaceConfig{
WorkspaceID: "ws-tenant",
PlatformURL: "http://localhost:8080",
Tier: 2,
EnvVars: map[string]string{
// Persona identity fields that are SAFE to keep (read-only
// identity, not a write credential):
"GITEA_USER": "backend-engineer",
"GITEA_USER_EMAIL": "backend-engineer@agents.moleculesai.app",
// The credential that must be stripped:
"GITEA_TOKEN": "persona-merged-write-pat",
"GITEA_TOKEN_SCOPES": "write:repository",
},
}
got := buildContainerEnv(cfg)
assertNoSCMWriteToken(t, got, scmTokens)
// Non-credential persona identity may still flow through — only the
// write token is the denied surface.
if !envContains(got, "GITEA_USER=backend-engineer") {
t.Errorf("non-credential persona identity (GITEA_USER) should not be stripped")
}
})
}
// TestCPProvisionerEnv_StripsSCMWriteTokens covers the tenant-EC2 path:
// CPProvisioner.Start builds the env map the control plane forwards to the
// EC2 workspace container. The same forensic #145 denylist must hold there.
func TestCPProvisionerEnv_StripsSCMWriteTokens(t *testing.T) {
// isSCMWriteTokenKey is the single source of truth shared by both
// buildContainerEnv (local Docker) and CPProvisioner.Start (tenant EC2).
// Assert it classifies every known SCM-write var as denied and leaves
// ordinary / read-only-identity vars alone.
for _, k := range []string{
"GITEA_TOKEN", "GITHUB_TOKEN", "GH_TOKEN",
"GITLAB_TOKEN", "GL_TOKEN", "BITBUCKET_TOKEN",
} {
if !isSCMWriteTokenKey(k) {
t.Errorf("isSCMWriteTokenKey(%q) = false, want true (SCM-write credential must be denied)", k)
}
}
for _, k := range []string{
"GITEA_USER", "GITEA_USER_EMAIL", "ANTHROPIC_API_KEY",
"CUSTOM", "PLATFORM_URL", "ADMIN_TOKEN", "",
} {
if isSCMWriteTokenKey(k) {
t.Errorf("isSCMWriteTokenKey(%q) = true, want false (must not over-strip non-SCM env)", k)
}
}
}
func assertNoSCMWriteToken(t *testing.T, env []string, scmTokens []string) {
t.Helper()
for _, e := range env {
key := e
if i := strings.IndexByte(e, '='); i >= 0 {
key = e[:i]
}
for _, banned := range scmTokens {
if key == banned {
t.Errorf("SCM-write credential %q leaked into workspace env (forensic #145 invariant violated): %q", banned, e)
}
}
}
}
func envContains(env []string, want string) bool {
for _, e := range env {
if e == want {
return true
}
}
return false
}
// ---------- buildWorkspaceMount — #65 workspace_access ----------
func TestBuildWorkspaceMount_SelectionMatrix(t *testing.T) {