testagent-transform: align with dummy-sink contract + static dev tokens (DEV-613)

The sink client in DEV-610 was built against the architecture-doc shape
(POST canonical bytes to `/api/v1/handover` with headers) but DEV-611's
dummy-sink speaks a different contract: `POST /api/v1/sink/dummy/deliver`
with a JSON wrapper `{decision_id, payload_hash, payload, approval_ref,
transform_ref}`. The two subtasks therefore never actually talked to each
other.

Reconcile on the dummy-sink's wire format because the sink's extra
fields (decision_id, approval_ref, transform_ref) are mandatory for the
verification-service replay guard and chain checks:

- Switch `sink.Client.Deliver` to take a `DeliverRequest` carrying all
  five fields, and parse the handover_ref / log_index out of the sink's
  response for the caller.
- Pull `decision_id` from the handover payload's top-level field inside
  `transform.Runner.Execute` and plumb it through to the sink, surface
  `SinkHandoverRef` on the result.
- Update the mock sink inside `runner_test.go` to accept the new wire
  format and assert every wrapper field survives round-trip.

Additionally add `TRANSFORM_DEV_POCKET_ID_TOKEN` /
`TRANSFORM_DEV_WORKLOAD_TOKEN` env vars so the Phase-1 smoke harness can
drive the agent without a Pocket-ID server or a projected SA token
mount; both short-circuit directly to a `StaticToken`.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

services/testagent-transform/README.md

@@ -33,10 +33,13 @@ to the Phase-1 dummy-sink. The exit code is 0 on success.
    `services/signing-service/internal/httpapi/sign.go`.
 5. **Log.** `POST {TRANSFORM_TLOG_URL}/api/v1/log/entries` with the
    envelope + `content_hash` from step 4.
-6. **Handover.** `POST {TRANSFORM_SINK_URL}/api/v1/handover` with the
-   canonical `handover_payload` bytes. Headers set:
-   - `X-Payload-Schema: sink.dummy.v1`
-   - `X-Attestation-Ref: <transform content_hash>`
+6. **Handover.** `POST {TRANSFORM_SINK_URL}/api/v1/sink/dummy/deliver`
+   with the DEV-611 wrapper JSON `{decision_id, payload_hash, payload,
+   approval_ref, transform_ref}`. The `payload` field is the canonical
+   handover_payload the agent already committed to in the transform
+   attestation's subject digest; the sink re-hashes and refuses the
+   delivery unless it matches `payload_hash`. The `decision_id` is read
+   from the top-level `decision_id` field of the handover payload.
 
 ## Non-scope
 
@@ -65,7 +68,9 @@ to the Phase-1 dummy-sink. The exit code is 0 on success.
 | `TRANSFORM_INPUT_SCHEMA` / `_VERSION`   | `expense.canonical` / `1.2.0`                              | Input schema name/version carried in the attestation.      |
 | `TRANSFORM_OUTPUT_SCHEMA` / `_VERSION`  | `sink.db.expense` / `1.2.0`                                | Output schema name/version carried in the attestation.     |
 | `TRANSFORM_VERSION`                  | `0.1.0`                                                       | Agent's own version string, carried in the attestation.    |
-| `TRANSFORM_SINK_PAYLOAD_SCHEMA`      | `sink.dummy.v1`                                               | Value sent on the sink's `X-Payload-Schema` header.        |
+| `TRANSFORM_SINK_PAYLOAD_SCHEMA`      | `sink.dummy.v1`                                               | Informational schema identifier recorded in the handover attestation. |
+| `TRANSFORM_DEV_POCKET_ID_TOKEN`      | *(unset)*                                                     | DEV/CI ONLY: static Pocket-ID bearer, bypasses the OIDC client-credentials grant. |
+| `TRANSFORM_DEV_WORKLOAD_TOKEN`       | *(unset)*                                                     | DEV/CI ONLY: static projected ServiceAccount token value.          |
 
 ## CLI
 

services/testagent-transform/cmd/testagent-transform/main.go

@@ -120,11 +120,18 @@ func loadConfig(keysEnv string) (runtimeConfig, error) {
 		return cfg, errors.New("TRANSFORM_TLOG_URL, TRANSFORM_SIGNING_URL, TRANSFORM_SINK_URL are required")
 	}
 
+	// DEV/CI short-circuits: if a static Pocket-ID token is provided, we
+	// skip the OIDC client-credentials dance entirely. Same for the
+	// projected SA token. Both are mutually exclusive with the real
+	// sources — set the static one or the trio of OIDC env vars.
+	if devPocketID := envRequired("TRANSFORM_DEV_POCKET_ID_TOKEN"); devPocketID != "" {
+		cfg.pocketIDSource = auth.StaticToken(devPocketID)
+	} else {
 		issuer := envRequired("POCKET_ID_ISSUER")
 		clientID := envRequired("POCKET_ID_CLIENT_ID")
 		clientSecret := envRequired("POCKET_ID_CLIENT_SECRET")
 		if issuer == "" || clientID == "" || clientSecret == "" {
-		return cfg, errors.New("POCKET_ID_ISSUER, POCKET_ID_CLIENT_ID, POCKET_ID_CLIENT_SECRET are required")
+			return cfg, errors.New("POCKET_ID_ISSUER, POCKET_ID_CLIENT_ID, POCKET_ID_CLIENT_SECRET are required (or set TRANSFORM_DEV_POCKET_ID_TOKEN for dev/CI)")
 		}
 		cfg.pocketIDSource = &auth.PocketIDTokenSource{
 			Issuer:       issuer,
@@ -132,9 +139,14 @@ func loadConfig(keysEnv string) (runtimeConfig, error) {
 			ClientSecret: clientSecret,
 			Audience:     envDefault("POCKET_ID_AUDIENCE", "trust-api"),
 		}
+	}
 
+	if devWorkload := envRequired("TRANSFORM_DEV_WORKLOAD_TOKEN"); devWorkload != "" {
+		cfg.workloadSource = auth.StaticToken(devWorkload)
+	} else {
 		saPath := envDefault("TRANSFORM_WORKLOAD_TOKEN_PATH", "/var/run/secrets/tokens/trust-api")
 		cfg.workloadSource = &auth.ProjectedSATokenSource{Path: saPath}
+	}
 
 	kr, err := loadKeyring(keysEnv)
 	if err != nil {

services/testagent-transform/internal/sink/client.go

@@ -1,14 +1,17 @@
-// Package sink posts a handover_payload to the (phase-1) dummy sink.
+// Package sink posts a handover envelope to the Phase-1 dummy sink.
 //
-// The payload is sent as the canonical JSON bytes the agent already hashed
-// into the transform attestation's subject digest — a byte-for-byte match
-// is a hard requirement for a downstream verifier, so we send those bytes
-// directly rather than re-marshaling.
+// The dummy-sink's contract (DEV-611) is `POST /api/v1/sink/dummy/deliver`
+// with a JSON wrapper `{decision_id, payload_hash, payload, approval_ref,
+// transform_ref}`. The sink re-hashes payload server-side and MUST see
+// `payload_hash` match either the raw bytes or the canonicalised form,
+// so we hand it the exact canonical bytes the transform attestation
+// already committed to.
 package sink
 
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -19,15 +22,6 @@ import (
 	"basicstack.de/proofedmesh/services/testagent-transform/internal/auth"
 )
 
-// SchemaHeader carries the payload's schema identifier so the sink knows
-// which decoder to pick. Phase-1 sinks are told to expect "sink.dummy.v1".
-const SchemaHeader = "X-Payload-Schema"
-
-// AttestationHeader carries the transform attestation's content_hash so
-// the sink can resolve the full attestation chain in the TLog without
-// re-parsing the signed envelope.
-const AttestationHeader = "X-Attestation-Ref"
-
 // ErrSinkUnavailable classifies connection errors and 5xx responses.
 var ErrSinkUnavailable = errors.New("sink: dummy-sink unavailable")
 
@@ -38,53 +32,100 @@ type Client struct {
 	TokenSource auth.TokenSource
 }
 
+// DeliverRequest carries everything the dummy-sink needs on its /deliver
+// endpoint. The runner builds this from the (approval_ref, transform_ref)
+// chain plus the canonical handover bytes.
+type DeliverRequest struct {
+	DecisionID   string
+	PayloadBytes []byte
+	PayloadHash  string
+	ApprovalRef  string
+	TransformRef string
+	Schema       string
+}
+
 // DeliverResult is what the agent reports upstream after a sink POST.
 type DeliverResult struct {
 	Status      int
+	HandoverRef string
+	LogIndex    int64
 }
 
-// Deliver POSTs payloadBytes to the sink. The caller passes the schema
-// identifier (e.g. "sink.dummy.v1") and the transform attestation's
-// content_hash so both go on the request headers.
-func (c Client) Deliver(ctx context.Context, schema, attestationRef string, payloadBytes []byte) (DeliverResult, error) {
+type deliverWire struct {
+	DecisionID   string          `json:"decision_id"`
+	PayloadHash  string          `json:"payload_hash"`
+	Payload      json.RawMessage `json:"payload"`
+	ApprovalRef  string          `json:"approval_ref"`
+	TransformRef string          `json:"transform_ref"`
+}
+
+type deliverResponseWire struct {
+	HandoverRef string `json:"handover_ref"`
+	LogIndex    int64  `json:"log_index"`
+	DecisionID  string `json:"decision_id"`
+}
+
+// Deliver POSTs the handover wrapper to the dummy-sink. The caller
+// passes the schema identifier (currently informational — the sink does
+// not read it) plus the canonical payload bytes committed to by the
+// transform attestation. The returned HandoverRef is the sink's own
+// handover attestation content_hash, surfaced so a caller can log it
+// alongside the transform ref for later chain inspection.
+func (c Client) Deliver(ctx context.Context, req DeliverRequest) (DeliverResult, error) {
 	if c.BaseURL == "" {
 		return DeliverResult{}, errors.New("sink.Client: BaseURL is empty")
 	}
-	if schema == "" {
-		return DeliverResult{}, errors.New("sink.Client: schema is required")
+	if req.DecisionID == "" {
+		return DeliverResult{}, errors.New("sink.Client: DecisionID is required")
 	}
-	if attestationRef == "" {
-		return DeliverResult{}, errors.New("sink.Client: attestationRef is required")
+	if len(req.PayloadBytes) == 0 {
+		return DeliverResult{}, errors.New("sink.Client: PayloadBytes is empty")
 	}
-	if len(payloadBytes) == 0 {
-		return DeliverResult{}, errors.New("sink.Client: payloadBytes is empty")
+	if req.PayloadHash == "" {
+		return DeliverResult{}, errors.New("sink.Client: PayloadHash is required")
 	}
+	if req.ApprovalRef == "" {
+		return DeliverResult{}, errors.New("sink.Client: ApprovalRef is required")
+	}
+	if req.TransformRef == "" {
+		return DeliverResult{}, errors.New("sink.Client: TransformRef is required")
+	}
+
 	hc := c.HTTPClient
 	if hc == nil {
 		hc = &http.Client{Timeout: 10 * time.Second}
 	}
-	endpoint, err := url.JoinPath(c.BaseURL, "/api/v1/handover")
+	endpoint, err := url.JoinPath(c.BaseURL, "/api/v1/sink/dummy/deliver")
 	if err != nil {
 		return DeliverResult{}, fmt.Errorf("sink: build url: %w", err)
 	}
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payloadBytes))
+	wire := deliverWire{
+		DecisionID:   req.DecisionID,
+		PayloadHash:  req.PayloadHash,
+		Payload:      json.RawMessage(req.PayloadBytes),
+		ApprovalRef:  req.ApprovalRef,
+		TransformRef: req.TransformRef,
+	}
+	body, err := json.Marshal(wire)
+	if err != nil {
+		return DeliverResult{}, fmt.Errorf("sink: marshal body: %w", err)
+	}
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
 	if err != nil {
 		return DeliverResult{}, err
 	}
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Accept", "application/json")
-	req.Header.Set(SchemaHeader, schema)
-	req.Header.Set(AttestationHeader, attestationRef)
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "application/json")
 	if c.TokenSource != nil {
 		tok, err := c.TokenSource.Token(ctx)
 		if err != nil {
 			return DeliverResult{}, fmt.Errorf("sink: token: %w", err)
 		}
 		if tok != "" {
-			req.Header.Set("Authorization", "Bearer "+tok)
+			httpReq.Header.Set("Authorization", "Bearer "+tok)
 		}
 	}
-	resp, err := hc.Do(req)
+	resp, err := hc.Do(httpReq)
 	if err != nil {
 		return DeliverResult{}, fmt.Errorf("%w: %v", ErrSinkUnavailable, err)
 	}
@@ -92,7 +133,15 @@ func (c Client) Deliver(ctx context.Context, schema, attestationRef string, payl
 	raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
 	switch {
 	case resp.StatusCode >= 200 && resp.StatusCode < 300:
-		return DeliverResult{Status: resp.StatusCode}, nil
+		var out deliverResponseWire
+		if err := json.Unmarshal(raw, &out); err != nil {
+			return DeliverResult{}, fmt.Errorf("sink: decode response: %w (body=%q)", err, truncate(raw, 256))
+		}
+		return DeliverResult{
+			Status:      resp.StatusCode,
+			HandoverRef: out.HandoverRef,
+			LogIndex:    out.LogIndex,
+		}, nil
 	case resp.StatusCode >= 500:
 		return DeliverResult{}, fmt.Errorf("%w: status %d: %s", ErrSinkUnavailable, resp.StatusCode, truncate(raw, 256))
 	default:

services/testagent-transform/internal/transform/runner.go

@@ -60,8 +60,9 @@ type Config struct {
 	// TransformVersion is the agent's own version string, carried into
 	// the attestation so bugfixes are auditable.
 	TransformVersion string
-	// SinkPayloadSchema is the schema identifier the sink expects on the
-	// `X-Payload-Schema` header. Phase-1: "sink.dummy.v1".
+	// SinkPayloadSchema is the schema identifier recorded in the
+	// handover attestation and forwarded on the sink.DeliverRequest.
+	// Phase-1: "sink.dummy.v1".
 	SinkPayloadSchema string
 }
 
@@ -81,9 +82,9 @@ type TLogSubmitter interface {
 	Submit(ctx context.Context, env dsse.Envelope, contentHash string) (tlog.Receipt, error)
 }
 
-// SinkDeliverer posts the handover_payload bytes to the sink.
+// SinkDeliverer posts the handover wrapper to the sink.
 type SinkDeliverer interface {
-	Deliver(ctx context.Context, schema, attestationRef string, payloadBytes []byte) (sink.DeliverResult, error)
+	Deliver(ctx context.Context, req sink.DeliverRequest) (sink.DeliverResult, error)
 }
 
 // Request is a single transform invocation. HandoverPayload is JSON-shaped
@@ -100,6 +101,7 @@ type Result struct {
 	TransformRef    string
 	LogIndex        int64
 	SinkStatus      int
+	SinkHandoverRef string
 	HandoverHashHex string
 }
 
@@ -153,7 +155,18 @@ func (r *Runner) Execute(ctx context.Context, req Request) (Result, error) {
 		return Result{}, fmt.Errorf("tlog submit transform: %w", err)
 	}
 
-	deliv, err := r.Sink.Deliver(ctx, r.SinkPayloadSchema, signed.ContentHash, payloadBytes)
+	decisionID, err := extractDecisionID(req.HandoverPayload)
+	if err != nil {
+		return Result{}, err
+	}
+	deliv, err := r.Sink.Deliver(ctx, sink.DeliverRequest{
+		DecisionID:   decisionID,
+		PayloadBytes: payloadBytes,
+		PayloadHash:  "sha256:" + payloadHashHex,
+		ApprovalRef:  req.ApprovalRef,
+		TransformRef: signed.ContentHash,
+		Schema:       r.SinkPayloadSchema,
+	})
 	if err != nil {
 		return Result{}, fmt.Errorf("sink deliver: %w", err)
 	}
@@ -162,10 +175,27 @@ func (r *Runner) Execute(ctx context.Context, req Request) (Result, error) {
 		TransformRef:    signed.ContentHash,
 		LogIndex:        receipt.LogIndex,
 		SinkStatus:      deliv.Status,
+		SinkHandoverRef: deliv.HandoverRef,
 		HandoverHashHex: payloadHashHex,
 	}, nil
 }
 
+// extractDecisionID pulls the decision_id string from a handover payload.
+// Phase-1 handover payloads carry it at the top level; the sink requires
+// it to key its (idempotent) write and the verification-service's replay
+// guard.
+func extractDecisionID(payload map[string]any) (string, error) {
+	v, ok := payload["decision_id"]
+	if !ok {
+		return "", errors.New("transform: handover_payload is missing top-level decision_id")
+	}
+	s, ok := v.(string)
+	if !ok || s == "" {
+		return "", errors.New("transform: handover_payload decision_id must be a non-empty string")
+	}
+	return s, nil
+}
+
 func (r *Runner) validate(req Request) error {
 	if r.Approvals == nil || r.ApprovalKeyring == nil || r.Signer == nil || r.TLog == nil || r.Sink == nil {
 		return errors.New("transform.Runner: missing dependency wiring")

services/testagent-transform/internal/transform/runner_test.go

@@ -9,7 +9,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -38,8 +37,10 @@ type mocks struct {
 	transformPriv ed25519.PrivateKey
 
 	sinkPayload      []byte
-	sinkSchema    string
-	sinkAttestRef string
+	sinkDecisionID   string
+	sinkPayloadHash  string
+	sinkApprovalRef  string
+	sinkTransformRef string
 	sinkCalled       bool
 
 	tlogSubmitted []string // sha256:… content_hashes seen on POST /log/entries
@@ -70,7 +71,7 @@ const (
 
 func buildMocks(t *testing.T, opts mockOpts) *mocks {
 	t.Helper()
-	m := &mocks{sinkSchema: "sink.dummy.v1"}
+	m := &mocks{}
 
 	// Keys.
 	approvalPub, approvalPriv, err := ed25519.GenerateKey(rand.Reader)
@@ -209,18 +210,38 @@ func buildMocks(t *testing.T, opts mockOpts) *mocks {
 		_ = json.NewEncoder(w).Encode(rcpt)
 	}))
 
-	// sinkSrv: POST /api/v1/handover; captures body + headers so the
-	// test can assert the agent sent exactly the canonical bytes.
+	// sinkSrv: POST /api/v1/sink/dummy/deliver with the DEV-611 wrapper
+	// shape; captures each field so the test can assert that the agent
+	// forwarded the canonical bytes verbatim and all four refs line up.
 	m.sinkSrv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/api/v1/handover" || r.Method != http.MethodPost {
+		if r.URL.Path != "/api/v1/sink/dummy/deliver" || r.Method != http.MethodPost {
 			http.NotFound(w, r)
 			return
 		}
 		m.sinkCalled = true
-		m.sinkSchema = r.Header.Get(sink.SchemaHeader)
-		m.sinkAttestRef = r.Header.Get(sink.AttestationHeader)
-		m.sinkPayload, _ = io.ReadAll(r.Body)
-		w.WriteHeader(http.StatusAccepted)
+		var body struct {
+			DecisionID   string          `json:"decision_id"`
+			PayloadHash  string          `json:"payload_hash"`
+			Payload      json.RawMessage `json:"payload"`
+			ApprovalRef  string          `json:"approval_ref"`
+			TransformRef string          `json:"transform_ref"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			http.Error(w, "decode: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+		m.sinkDecisionID = body.DecisionID
+		m.sinkPayloadHash = body.PayloadHash
+		m.sinkPayload = append(m.sinkPayload[:0], body.Payload...)
+		m.sinkApprovalRef = body.ApprovalRef
+		m.sinkTransformRef = body.TransformRef
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"handover_ref": "sha256:" + strings.Repeat("d", 64),
+			"log_index":    int64(99),
+			"decision_id":  body.DecisionID,
+		})
 	}))
 
 	t.Cleanup(func() {
@@ -269,6 +290,7 @@ func sampleRequest() Request {
 		WorkflowID:  "WF-TEST-1",
 		ApprovalRef: attestationRef,
 		HandoverPayload: map[string]any{
+			"decision_id": "dec-WF-TEST-1",
 			"schema":      map[string]any{"name": "sink.dummy.v1", "version": "1.0.0"},
 			"items": []any{
 				map[string]any{"line": "1", "amount": "12.34"},
@@ -304,11 +326,20 @@ func TestExecute_HappyPath(t *testing.T) {
 	if !bytes.Equal(m.sinkPayload, wantBytes) {
 		t.Fatalf("sink payload bytes differ:\n got %q\nwant %q", m.sinkPayload, wantBytes)
 	}
-	if m.sinkSchema != "sink.dummy.v1" {
-		t.Fatalf("sink schema header = %q", m.sinkSchema)
+	if m.sinkDecisionID != "dec-WF-TEST-1" {
+		t.Fatalf("sink decision_id = %q, want %q", m.sinkDecisionID, "dec-WF-TEST-1")
 	}
-	if m.sinkAttestRef != res.TransformRef {
-		t.Fatalf("sink attestation header = %q, want %q", m.sinkAttestRef, res.TransformRef)
+	if m.sinkPayloadHash != "sha256:"+wantHash {
+		t.Fatalf("sink payload_hash = %q, want %q", m.sinkPayloadHash, "sha256:"+wantHash)
+	}
+	if m.sinkApprovalRef != req.ApprovalRef {
+		t.Fatalf("sink approval_ref = %q, want %q", m.sinkApprovalRef, req.ApprovalRef)
+	}
+	if m.sinkTransformRef != res.TransformRef {
+		t.Fatalf("sink transform_ref = %q, want %q", m.sinkTransformRef, res.TransformRef)
+	}
+	if res.SinkHandoverRef == "" {
+		t.Fatalf("SinkHandoverRef unset — expected sink's handover_ref")
 	}
 
 	// TLog must have seen exactly our transform content_hash.
@@ -318,8 +349,8 @@ func TestExecute_HappyPath(t *testing.T) {
 	if res.LogIndex == 0 {
 		t.Fatalf("LogIndex not set")
 	}
-	if res.SinkStatus != http.StatusAccepted {
-		t.Fatalf("SinkStatus = %d, want 202", res.SinkStatus)
+	if res.SinkStatus != http.StatusCreated {
+		t.Fatalf("SinkStatus = %d, want 201", res.SinkStatus)
 	}
 }