Cognitive Services Pipeline API

This REST facade exists so the frontend can drive the pipeline over plain JSON without a gRPC client. It is an interim REST facade in the Assentify Orchestrator (Assentify.ML.Face.SignalR) that forwards to the AI Interface. Endpoints and payload shapes are stable; the transport may later move to a dedicated Assentify.CognitiveServices service without changing this contract.

Three POST endpoints, all under /api/Pipeline. Pick by how much you want to happen:

All calls target the Orchestrator base URL:

{ORCHESTRATOR_BASE_URL}/api/Pipeline/...

Content type is application/json for all requests. Validate and DryRun return JSON; Execute streams application/x-ndjson. All field names are camelCase, both request and response.

CORS is already whitelisted for localhost:4200 (your Angular dev server) and the *.platform.assentify.com / *.sls.assentify.com origins.

Used by DryRun and Execute. Sent as an int.

Optional; available on DryRun and Execute (batch-level, applied to every image). All default off / AUTO.

1. Send raw base64 or a data: URI (data:image/jpeg;base64,…). The server strips the data:…, prefix and any whitespace or newlines before processing.
2. The per-image size cap is measured on the cleaned (decoded) payload.
3. Invalid base64 (bad length) makes that image's line ok:false, errorType:"validation".

Validate pipeline YAML, optionally in the context of a template.

Request

POST {BASE}/api/Pipeline/Validate
Content-Type: application/json

{
  "pipelineYaml": "steps:\n  - detect\n  - ocr\n  - validate",
  "templateId": "8e0f...c1"
}

Response: 200 ValidatePipelineResponse

{
  "valid": false,
  "errors": [{ "code": "...", "message": "...", "path": "..." }],
  "warnings": [{ "code": "...", "message": "...", "path": "..." }]
}

Resolve and plan a pipeline for one document without executing it. Returns execution order, per-step plans, estimated cost, the resolved YAML, and the single status the request would resolve to. The frontend supplies the template (referenceId) and the image.

Request

POST {BASE}/api/Pipeline/DryRun
Content-Type: application/json

{
  "referenceId": "8e0f...c1",
  "imageFile": "/9j/4AAQSk...",
  "documentType": 1,
  "documentTampering": true,
  "ocrType": 1
}

Response: 200 DryRunPipelineResponse

{
  "valid": true,
  "executionOrder": ["detect", "ocr", "validate"],
  "stepPlans": [
    { "id": "detect", "phase": "...", "conditional": false,
      "abortStatus": "DetectIDFailed", "onFailure": "ABORT", "...": "..." }
  ],
  "estimatedCost": { "...": "..." },
  "resolvedPipelineYaml": "steps:\n  ...",
  "errors": [],
  "status": 210
}

Execute a pipeline against one or more images. Images are processed with bounded concurrency and each result is streamed back as it completes, one NDJSON object per line, so the UI can render progress per image.

Request body: ExecutePipelineBatchDto

Each image item (images[]):

Request

POST {BASE}/api/Pipeline/Execute
Content-Type: application/json
Accept: application/x-ndjson

{
  "referenceId": "8e0f...c1",
  "documentType": 1,
  "documentTampering": true,
  "params": { "locale": "ar" },
  "images": [
    { "imageFile": "/9j/4AAQSk...", "correlationId": "front" },
    { "imageFile": "/9j/4AAQSk...", "correlationId": "back" }
  ]
}

Upload limits (this endpoint)

Streaming response (application/x-ndjson)

HTTP 200, then one JSON object per line (\n-delimited). Lines arrive in completion order, not submission order. Use index or correlationId to attribute them.

// one object per line, flushed as each image finishes
{ "index":0, "correlationId":"front", "ok":true, "response":{ ...ExecutePipelineResponse... }}
{ "index":1, "correlationId":"back", "ok":false, "errorType":"transport", "error":"deadline exceeded" }

Each line (StreamedPipelineResult):

response (when ok:true): the full ExecutePipelineResponse

{
  "success": true,
  "status": 210,
  "error": "",
  "response": {
    "scannedDoc": "...", "faces": ["..."], "croppedCard": "...", "croppedMrz": "...",
    "classifiedTemplate": "...", "rawResult": "...",
    "zonalResult": { "...": "..." },
    "finalFields": { "...": "..." },
    "fields": { "...": "..." },
    "mrzFields": { "...": "..." },
    "documentAuth": true, "isAllowed": true, "flag": false, "documentTampering": false,
    "faceLive": true, "croppedFace": "...", "faceEncoding": "...",
    "digitalSignature": "...", "decodedData": "...", "decodedFace": "...", "decodedSignature": "...",
    "heatmap": "...", "tamperingPercentage": 0
  }
}

Error cases (before streaming starts)

The wrapper below turns the NDJSON stream into an Observable<StreamedPipelineResult> that emits one line at a time and aborts the request on unsubscribe.

// pipeline-stream.ts
import { Observable } from "rxjs";

export function executePipeline$(
  baseUrl: string,
  body: ExecutePipelineBatchDto,
): Observable<StreamedPipelineResult> {
  return new Observable<StreamedPipelineResult>((subscriber) => {
    const controller = new AbortController();

    (async () => {
      const res = await fetch(`${baseUrl}/api/Pipeline/Execute`, {
        method: "POST",
        headers: { "Content-Type": "application/json", Accept: "application/x-ndjson" },
        body: JSON.stringify(body),
        signal: controller.signal,
      });

      // Pre-stream errors (400 too many/empty, 413 too large) arrive as a single JSON body.
      if (!res.ok || !res.headers.get("content-type")?.includes("application/x-ndjson")) {
        const problem = await res.json().catch(() => ({}));
        throw new Error(`Execute failed (${res.status}): ${problem.detail ?? problem.title ?? res.statusText}`);
      }

      const reader = res.body!.getReader();
      const decoder = new TextDecoder();
      let buffer = "";

      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });

        let nl: number;
        while ((nl = buffer.indexOf("\n")) >= 0) { // emit each complete line
          const line = buffer.slice(0, nl).trim();
          buffer = buffer.slice(nl + 1);
          if (line) subscriber.next(JSON.parse(line) as StreamedPipelineResult);
        }
      }
      const tail = buffer.trim(); // flush any final partial line
      if (tail) subscriber.next(JSON.parse(tail) as StreamedPipelineResult);
    })()
      .then(() => subscriber.complete())
      .catch((err) => { if (!controller.signal.aborted) subscriber.error(err); });

    return () => controller.abort(); // unsubscribe -> abort the request
  });
}

Component usage, attribute each line by index / correlationId and branch on errorType:

import { Component, inject, signal } from "@angular/core";
import { takeUntilDestroyed } from "@angular/core/rxjs-interop";
import { environment } from "./environment";

@Component({ /* ... */ })
export class BatchRunnerComponent {
  private destroyRef = inject(DestroyRef);
  readonly results = signal<Record<string, StreamedPipelineResult>>({});

  run(files: string[]): void {
    const body: ExecutePipelineBatchDto = {
      referenceId: this.templateId,
      documentType: 1,
      documentTampering: true,
      images: files.map((b64, i) => ({ imageFile: b64, correlationId: `img-${i}` })),
    };

    executePipeline$(environment.orchestratorBaseUrl, body)
      .pipe(takeUntilDestroyed(this.destroyRef)) // leaving the view aborts the request
      .subscribe({
        next: (line) => {
          this.results.update((r) => ({ ...r, [line.correlationId]: line }));
          if (line.ok) {
            this.applyResult(line.correlationId, line.response!.response); // note: .response.response
          } else if (line.errorType === "validation") {
            this.showBadInput(line.correlationId, line.error!); // do NOT retry
          } else {
            this.showRetryable(line.correlationId, line.error!); // transport/error -> retryable
          }
        },
        error: (err) => this.toast(String(err.message ?? err)), // pre-stream 400/413
      });
  }
}