custom_python / custom_python_preprocess
code + requirements params] + NDW[Node_Designer CreateWizard / GeneratePanel
frame_processing_hook.py tab] + CAP[CodeAssistPanel
prompt, review, accept/reject
pure reducer codeAssistState.ts] + IA[importAnalyzer.ts
extractImports -> deriveRequirements
-> reconcileRequirements
IMPORT_MAPPING + STDLIB tables] + API[services/api.ts
codeAssist request] + NCP --> CAP + NDW --> CAP + NCP -->|debounced code change| IA + IA -->|reconciled requirements string| NCP + CAP --> API + end + subgraph Backend[WorkflowGeneratorHandler Lambda] + RT[workflow_generator.handler
route dispatch] + CA[code_assist.py
RBAC per surface, prompt assembly,
invocation, entry-point validation] + BC[bedrock_common.py shared module
get_bedrock_configuration
get_bedrock_client, defaults] + WG[workflow_generator.py
existing /workflows/generate] + RT --> CA + RT --> WG + CA --> BC + WG --> BC + end + API -->|POST /code-assist| APIGW[API Gateway] + APIGW --> RT + CA -->|Converse, forced tool use| BR[Amazon Bedrock] + BC -->|bedrock_configuration| DDB[(Settings table)] + CA -->|denied attempts| AUD[(Audit log)] +``` + +Request flow for one assist: + +1. The user types a prompt (1–4,000 chars) in the CodeAssistPanel and submits. The panel reducer enters `submitting` (spinner shown, submit disabled, editor untouched and still editable). +2. The frontend POSTs `/code-assist` with `{usecase_id, surface, contract, prompt, current_code, context}` — `current_code` is sent only when the editor holds a non-whitespace character (Requirements 2.6, 2.10). +3. `code_assist.py` authorizes per surface (workflow permission or Node_Designer role), returning 403 + audit entry before any generation on denial (Requirement 6.3). +4. It loads the Bedrock_Configuration through the shared module, assembles the contract-specific system prompt and user message, and invokes Converse with a forced `provide_code` tool call. +5. The tool output's `code` is validated server-side with `ast.parse` + entry-point inspection; failures return categorized error envelopes; success returns `{code, notes, model_id}`. +6. The panel shows the returned code for review; Accept writes it into the editor (which triggers the Import_Analyzer on Workflow_Builder surfaces), Reject discards it and keeps the prompt. +7. Independently of the assistant, every editor code change (typed or accepted) runs the Import_Analyzer after a 750 ms debounce — well inside the 2-second bound of Requirement 3.1 — and reconciles the node's `requirements` parameter. + +### Key Design Decisions + +| Decision | Choice | Rationale | +|---|---|---| +| Where the Import_Analyzer runs | Frontend TypeScript, pure synchronous module | Requirement 3.1's 2-second reactive bound applies to *every* keystroke-driven code change; an in-browser pure function is instant, works offline from the backend, adds no per-edit Lambda traffic, and — being pure — is the feature's best property-test target. The mapping table is **not shared with the backend** because the backend never derives requirements (generation returns code only; derivation always happens editor-side, uniformly for typed and generated code, Requirement 3.1). One table, one owner, no cross-language sync problem. | +| Import extraction technique | Hand-written line/continuation-aware scanner over import statements (comments and string literals stripped first), not a full Python parser | A real Python grammar in the browser (pyodide, tree-sitter WASM) costs megabytes for one function. The scanner recognizes the complete `import` / `from … import` statement grammar including parenthesized and backslash continuations and nested (indented) imports. "Cannot be parsed" (Requirement 3.10) is defined as: a malformed import statement or an unterminated string/paren encountered while scanning — precisely the failures that could corrupt derivation. Syntax errors *outside* import statements cannot change the import set, so deriving from a module that is otherwise broken is safe and useful (the user is mid-edit most of the time). | +| Derived-vs-manual encoding | Trailing marker comment on derived lines: `name # via code imports` (unmapped: `name # via code imports (verify package name)`) | The `requirements` parameter string is the only persisted artifact (packaged verbatim as requirements.txt, where comments are legal). A sidecar data structure would desync on reload/duplicate/generate paths. Any line **without** the marker is manual and preserved verbatim — which automatically protects user pins like `numpy==1.24.0` (Requirements 3.5, 3.9). | +| `cv2` mapping | `opencv-python-headless` | Matches the catalog's own `requirements` example and the edge LocalServer's installed OpenCV distribution; `opencv-python` would drag GUI/X11 dependencies onto headless devices. | +| Endpoint shape | Synchronous `POST /code-assist` served by the existing WorkflowGeneratorHandler Lambda (new `code_assist.py` module, dispatched from `workflow_generator.handler`) | Single-file code generation is a small completion (one module, ≤ `max_tokens`), far quicker than the 45–50 s multi-file scaffold generation that forced `node_generator.py` into start/poll sessions. Reusing the Lambda avoids a new function/role/permission set in a nested stack near the CloudFormation resource limit, and the Lambda already holds Bedrock invoke permissions and the settings-table read. The API Gateway 29 s integration ceiling is a pre-existing, shared caveat with `POST /workflows/generate`; a gateway-cut request surfaces through the same timeout error path (Requirement 5.2). | +| Statelessness | No chat sessions; each request carries the current editor code | Requirement 2.6 defines follow-up semantics purely in terms of the current editor content. Dropping sessions removes the DynamoDB table, TTL, and S3 snapshots the workflow generator needs, and guarantees Requirement 6.4 (authorization evaluated fresh per request) structurally. | +| Structured output | Converse forced tool use (`provide_code` tool with `{code, notes}` input schema) | Same mechanism as both existing generators: extraction is a field read, never regex-scraping markdown fences. "No tool call or empty `code`" is the well-defined trigger for Requirement 5.3. | +| Entry-point validation location | Backend, `ast.parse` + top-level `FunctionDef` inspection | The Lambda has a real Python parser; the response is then either code that provably parses and carries the contract's entry point, or a categorized error (Requirements 2.2, 2.3, 5.6). For `custom_python` the validator requires **exactly one** of `process_frame`/`handle` (Requirement 2.3) — two entry points would silently shadow one another at runtime (the bridge prefers `process_frame`). | +| Shared config module | New `backend/functions/bedrock_common.py`; `workflow_generator.py` refactored to import it; `node_generator.py` migration optional follow-up | Requirement 4 pins the assist feature to workflow generation's exact semantics; sharing code makes divergence impossible. The `node_generator.py` copy is left as-is to keep this feature's diff contained (its docstrings/session logic have diverged); migrating it is a mechanical cleanup, not a requirement. | + +## Components and Interfaces + +### 1. `bedrock_common.py` — shared Bedrock configuration (backend/functions) + +Extracted verbatim from `workflow_generator.py` (Requirements 4.1–4.7): + +```python +BEDROCK_CONFIG_SETTING_KEY = 'bedrock_configuration' +MAX_TIMEOUT_SECONDS = 60 +DEFAULT_BEDROCK_CONFIG = { ... } # unchanged values + +def get_bedrock_configuration() -> Dict: # unchanged semantics: + # defaults -> stored overrides; temperature/top_p honor explicit null; + # timeout coerced to int, junk -> 60, clamped to [1, 60] +def get_bedrock_client(region: str, timeout_seconds: int): # unchanged: + # per-(region, timeout) cache, connect_timeout min(t, 10), + # read_timeout = t, retries disabled +def build_inference_config(config: Dict) -> Dict: + # {'maxTokens': int(config['max_tokens'])} plus at most ONE sampling + # parameter: temperature when set, else topP when set (4.2, 4.3) +``` + +`workflow_generator.py` replaces its local copies with imports (same Lambda bundle — `backend/functions` is one code asset — so this is a same-directory import exactly like its existing `from workflow_validation import …`). `build_inference_config` is newly factored out of `invoke_generation` so both callers share the sampling-exclusivity rule. + +### 2. `code_assist.py` — Code_Assist_Generator (backend/functions) + +Handles `POST /code-assist`, dispatched from `workflow_generator.handler` (the module lives in the same bundle; the handler gains one `resource == '/code-assist'` branch). Same error envelope, `parse_body`, `forbidden_response`-style helpers as the existing generators. + +**Request validation** (400 on failure): +- `usecase_id`, `surface`, `contract`, `prompt` required (`MISSING_FIELDS`) +- `surface` ∈ {`workflow-builder`, `node-designer`} (`INVALID_SURFACE`) +- `contract` ∈ {`process_frame`, `process_frame_or_handle`, `frame_hook`} (`INVALID_CONTRACT`) +- `prompt` a string with ≥ 1 non-whitespace character and `len(prompt) <= 4000` (`INVALID_PROMPT`) — the server-side twin of the frontend check (Requirements 1.4, 2.8) +- `current_code` optional string; `context` optional object (`{node_type?, parameters?: [{name, param_type, description?}]}`, used by `frame_hook` prompts) + +**Authorization** (Requirements 6.1–6.4), evaluated on every request before any Bedrock call: + +```python +def is_authorized(user, usecase_id, surface) -> bool: + if surface == 'workflow-builder': + return has_permission(WORKFLOW_CREATE) or has_permission(WORKFLOW_EDIT) + return rbac_manager.get_user_role(user, usecase_id) == 'UseCaseAdmin' \ + or user_is_portal_admin(user) # same rule as node_generator.can_generate +``` + +Denial returns the uniform 403 `FORBIDDEN` envelope and writes an `unauthorized_access` audit entry carrying the acting user, `surface`, `usecase_id`, and timestamp (Requirement 6.3) — the same `log_audit_event` call shape as `forbidden_response` in `workflow_generator.py`. `get_usecase(usecase_id)` failure returns 404 `USECASE_NOT_FOUND`. + +**Prompt assembly** — pure functions (property-tested): + +```python +CONTRACTS = { + 'process_frame': { + 'entry_points': frozenset({'process_frame'}), 'require_exactly_one': False, + 'signature': 'process_frame(frame, metadata)', + 'environment': PYTHON_BRIDGE_ENVIRONMENT, # see below + }, + 'process_frame_or_handle': { + 'entry_points': frozenset({'process_frame', 'handle'}), 'require_exactly_one': True, + 'signature': 'process_frame(frame, metadata) or handle(frame_bytes, metadata)', + 'environment': PYTHON_BRIDGE_ENVIRONMENT, + }, + 'frame_hook': { + 'entry_points': frozenset({'process_frame'}), 'require_exactly_one': False, + 'signature': 'process_frame(frame, params)', + 'environment': FRAME_HOOK_ENVIRONMENT, + }, +} + +def build_system_prompt(contract: str, context: Optional[Dict]) -> str +def build_user_message(prompt: str, current_code: Optional[str]) -> str +``` + +`PYTHON_BRIDGE_ENVIRONMENT` describes the Python_Bridge runtime faithfully (Requirement 2.1), sourced from `python_bridge.py`: +- `process_frame(frame, metadata)`: `frame` is a NumPy uint8 array (H×W×C; H×W for GRAY8, formats RGB/BGR/RGBA/GRAY8); return `None` to pass the frame through, or an array of **identical shape and dtype** (the bridge rejects anything else); mutate `metadata` in place to attach results. +- `handle(frame_bytes, metadata)`: raw bytes in, `(frame_bytes, metadata)` out. +- `cv2`, `np`, and `numpy` are pre-bound on the module — no import needed, but an explicit import is harmless. +- `import dda_frames` provides `to_array(frame_bytes, width, height, format)`, `to_bytes(array)`, `frame_info()` → `{'width', 'height', 'format'}`, `load_image(path_or_s3_uri)` (BGR uint8). +- `metadata["frame"]` carries `{width, height, format}` on every invocation. +- Never write to stdout (it belongs to the frame protocol); use `sys.stderr` for diagnostics. +- Extra pip packages may be imported freely; the Portal derives the node's requirements from the imports (so the model should emit a normal `import` for any library the user asks for — Requirement 3.4). +- Keep the module complete and self-contained: when current code is provided, return the **entire modified module**, never a fragment or diff (Requirement 2.6). + +`FRAME_HOOK_ENVIRONMENT` describes the Frame_Processing_Hook: `process_frame(frame, params)` with `params` holding the declared element parameters (names/types injected from `context.parameters`), embedded-interpreter execution, return the processed frame. + +`build_user_message(prompt, current_code)` embeds `current_code` in a `CURRENT MODULE CODE` block with modify-not-regenerate instructions **iff** it contains a non-whitespace character; otherwise it sends the prompt alone (Requirements 2.6, 2.10) — mirroring `workflow_generator.build_user_message`. + +**Invocation** — `get_bedrock_client(...).converse(...)` with `build_inference_config(config)` and forced tool use: + +```python +TOOL_NAME = 'provide_code' +# inputSchema: {type: object, required: [code], properties: { +# code: {type: string, description: 'the complete Python module'}, +# notes: {type: string, description: 'one short paragraph for the user'}}} +``` + +Exception mapping mirrors `invoke_generation` and adds Requirement 5.1's categories in `details.category` (see Error Handling). + +**Output validation** — pure function (property-tested): + +```python +def validate_entry_point(code: str, contract: str) -> Optional[str]: + """None when valid; a defect description otherwise. + - ast.parse failure -> 'generated code is not valid Python: ...' + - top-level FunctionDef names are intersected with the contract's + entry_points; zero matches -> 'missing entry point ...' + - require_exactly_one and both + process_frame and handle defined -> 'defines both entry points ...' + """ +``` + +Parse failure returns 422 `GENERATED_CODE_INVALID`; a missing/duplicated entry point returns 422 `MISSING_ENTRY_POINT` (Requirements 2.2, 2.3, 5.6). Empty/absent tool output returns 422 `NO_CODE_RETURNED` (Requirement 5.3). + +**Success response** (200): + +```json +{ "code": "...", "notes": "...", "model_id": "us.anthropic....", "contract": "process_frame" } +``` + +Nothing is persisted anywhere — no DynamoDB, no S3 (Requirements 2.7, 6.4). + +### 3. Infrastructure (api-gateway-stack.ts, compute-stack.ts) + +- `api-gateway-stack.ts`: one new top-level resource `code-assist` with `POST` on the existing `workflowGeneratorIntegration` (Cognito authorizer, `allowTestInvoke: false`, CORS OPTIONS like its siblings). Two CloudFormation resources total; no new Lambda. +- `compute-stack.ts`: no change needed — `WorkflowGeneratorHandler` already bundles `backend/functions`, has the 60 s timeout, the settings-table read, and `bedrock:InvokeModel*` permissions used by `/workflows/generate`. + +### 4. `importAnalyzer.ts` — Import_Analyzer (frontend/src/pages/workflows/) + +A dependency-free pure module (the feature's main property-test target): + +```typescript +/** Result of scanning module code for import statements. */ +export type ImportScan = + | { ok: true; imports: string[] } // absolute top-level module names, deduped + | { ok: false }; // unparseable (Requirement 3.10) + +export function extractImports(code: string): ImportScan; +``` + +The scanner strips comments and string literals (tracking single/double/triple quotes; an unterminated string ⇒ `{ok:false}`), joins backslash and open-paren continuations, and matches every logical line — at any indentation, so nested imports count (Requirement 3.1) — against the import-statement grammar: + +- `import a.b.c as x, d` → top-level names `a`, `d` +- `from a.b import x, y` → top-level name `a` +- `from . import x` / `from .sib import x` → **relative**: recorded as excluded (Requirement 3.3) +- A line starting with `import`/`from` that does not match the grammar ⇒ `{ok:false}` + +```typescript +/** One derived requirements entry. */ +export interface DerivedRequirement { + distribution: string; // pip distribution name + needsReview: boolean; // true when the import had no mapping (3.7) +} + +export function deriveRequirements(imports: string[]): DerivedRequirement[]; +``` + +`deriveRequirements` drops names in `STDLIB_MODULES` (the union of CPython 3.9 and 3.11 `sys.stdlib_module_names`, plus `__future__`) and `dda_frames`, then maps the rest: `IMPORT_MAPPING[name]` when present (`needsReview: false`), else the import name itself with `needsReview: true` (Requirements 3.2, 3.3, 3.7). Output is sorted and deduped. + +```typescript +export const IMPORT_MAPPING: Record
Node_Designer pages] + APIGW[API Gateway] + PR[plugin_records.py Lambda
records, lifecycle, review] + IMP[plugin_importer.py Lambda
repo + Module_Listing import] + GEN2[node_generator.py Lambda
Bedrock scaffold generation] + BLD[plugin_builds.py Lambda
build orchestration] + CB[CodeBuild projects
per Target_Architecture
incl. x86_64_nvidia CUDA image] + PPKG[plugin_components.py Lambda
Plugin_Component auto-packaging
dda.plugin.pluginId] + SIM[Step Functions +
Fargate sandbox
Plugin_Simulator] + KMS[KMS signing key] + DDB[(DynamoDB
PluginRecords, CustomNodeTypes,
ModuleIndexCache, SimulationRuns)] + S3P[(Portal S3
plugin sources, scaffolds,
Plugin_Library, sim results)] + PKG[workflow_packaging.py
Component_Packager - extended
declares Plugin_Component deps] + DEP[deployments.py + components.py
Plugin_Component listing +
lifecycle/arch deployment gates] + WV[workflow_validation.py +
workflow_generator.py
merged catalog consumers] + end + GGR[(Greengrass registry
Use_Case account
dda.workflow.* + dda.plugin.*)] + EXT[Public repositories +
gstreamer.freedesktop.org/modules/] + BR[Amazon Bedrock] + subgraph Edge Device + LS[LocalServer
checksum-verified plugin load] + end + + FE --> APIGW + APIGW --> PR & IMP & GEN2 & BLD & DEP + IMP --> EXT + IMP --> DDB + GEN2 --> BR + BLD --> CB + CB --> S3P + CB --> KMS + BLD --> DDB + BLD -.build success event.-> PPKG + PPKG --> S3P + PPKG --> GGR + PPKG --> DDB + APIGW --> SIM + SIM --> S3P + PR --> DDB + PKG --> S3P + PKG --> KMS + PKG --> GGR + DEP --> GGR + WV --> DDB + GGR -.installs Plugin_Components +
Workflow_Components.-> LS +``` + +### The Plugin_Record pipeline + +Every path produces the same artifact flow: + +```mermaid +stateDiagram-v2 + [*] --> dev : create / generate / import
(review = pending) + dev --> dev : edit source, rebuild, simulate + dev --> test : UseCaseAdmin promote
guard - at least one built Plugin_Artifact + test --> prod : UseCaseAdmin promote
guard - approved security review + test --> dev : demote + prod --> test : demote + note right of test : palette visible with test marker
cloud test runs allowed
deploy to Test_Devices only + note right of prod : deployable to any device +``` + +New source, changed declaration, or a rebuild creates a **new version item** whose Lifecycle_State is `dev` and review decision `pending`, independent of prior versions (Requirements 9.13, 10.5). Deployed Workflow_Components are never touched by demotion — gates apply only to subsequent packaging/deployment requests (9.12). + +### Dynamic Node_Type_Catalog extension + +The static catalog stays exactly as it is. A new module `workflow_core.catalog.custom` adds: + +- `descriptor_from_declaration(decl: dict) -> NodeTypeDescriptor` — converts a stored Custom_Node_Type declaration (same wire shape the node-catalog endpoint already serves) into a frozen `NodeTypeDescriptor`, validating port types against `PORT_TYPES`, categories against `CATEGORIES`, parameter descriptors against `PARAMETER_TYPES`, and mappings against `ARCHITECTURES`. Invalid declarations raise `DeclarationError` identifying the offending field (Requirement 8.5). +- `resolve_catalog(custom_descriptors: Sequence[NodeTypeDescriptor]) -> tuple` — returns `NODE_CATALOG + tuple(custom_descriptors)` with duplicate `type_id` rejection (built-ins always win). + +Consumers change as follows (all portal-side; the sandbox and LocalServer vendored copies gain the module but the edge never resolves custom catalogs — compiled documents are self-contained): + +| Consumer | Change | +|---|---| +| `GET /workflows/node-catalog` | Loads the Use_Case's registered Custom_Node_Types (Lifecycle_State test/prod only; dev excluded per 9.2), merges, serves; test-state entries carry `lifecycleState: "test"` for the palette marker (9.6) | +| `workflow_validation.py` | Validates against the merged catalog for the workflow's Use_Case | +| `workflow_generator.py` | Serializes the merged catalog into the system prompt so generated workflows may use registered custom nodes | +| `workflow_packaging.py` | Compiles against the merged catalog; custom-node plugin dependencies resolve to Plugin_Component dependencies declared in the Workflow_Component recipe, with artifact verification against the Plugin_Record (below) | +| test runner (`workflow_test_steps.py`) | Compile step uses the merged catalog; the sandbox task downloads custom x86_64 Plugin_Artifacts into its plugin scan path (Requirement 12.1) | + +A custom `GstMapping` for a plugin-backed node is mechanical: `element_chain = [{factory:
image = arch build image, env = arch) + CB->>CB: meson/autotools build in ephemeral container
(no cross-Use_Case credentials or data) + CB->>S3: upload .so to staging key + CB->>KMS: Sign(SHA-256 digest) + CB->>S3: promote to Plugin_Library custom/{usecase}/{arch}/ + CB-->>BLD: build result webhook (EventBridge) + BLD->>DDB: record artifact {s3Key, checksum, signature, status} + end + BLD->>BLD: all requested arch builds settled +
at least one succeeded + BLD-)PPKG: plugin_components.py: package Plugin_Component
dda.plugin.{pluginId} v{pluginVersion}.0.0 (Requirement 16.1) + BLD-->>UI: per-arch build status (succeeded/failed + log excerpt) +``` + +- **Isolation (3.2)**: each build runs in a fresh CodeBuild container with a role scoped to exactly the source prefix and staging prefix of that build — no Use_Case account credentials, no other builds' prefixes. Build projects run without VPC access to portal internals; outbound internet is allowed for source-declared dependency fetches during import builds. +- **Failure (3.4)**: failed builds store the CloudWatch log tail in the Plugin_Record's per-arch entry; no artifact is stored for that arch. +- **Prebuilt binaries (3.6)**: an upload path accepts a `.so` per arch, checksums and signs it identically (provenance records `prebuilt: true`). +- **DeepStream (5.1, 5.2)**: records flagged `deepstream: true` restrict selectable architectures to `arm64_jp4/jp5/jp6`; the JetPack build images pin the DeepStream SDK version matching each JetPack release. + +### Plugin_Scaffold and Node_Generator + +- Scaffold generation is pure templating in a new `workflow_core.scaffold` module (also usable in tests without AWS): given a validated declaration (name, category, ports, parameters, architectures), it renders a GStreamer plugin project — a C skeleton element wrapping an embedded Python `process_frame(frame, params) -> frame` Frame_Processing_Hook file (the same appsink/appsrc bridge approach as the existing `emlpython` custom-python element), `meson.build` per selected architecture, and a README. Declared parameters surface as GObject properties plumbed into the hook's `params` dict (Requirements 1.2–1.4). The scaffold is stored under portal S3 `plugin-sources/{usecase_id}/{plugin_id}/{version}/` and downloadable as a zip (1.5). +- The Node_Generator (`node_generator.py`) mirrors `workflow_generator.py`: chat sessions in a TTL'd DynamoDB table with the current scaffold source snapshot in S3; Converse invocation with a forced `create_plugin_scaffold` tool whose input schema is the scaffold file map (`{files: {path: content}}`) plus the declaration; the system prompt embeds the scaffold template conventions and the Frame_Processing_Hook contract. Follow-up prompts include the current source and instruct modification (2.4). Output that fails scaffold validation (missing hook file, undeclared files) returns an error with the prompt preserved (2.6); Bedrock failures/timeouts follow the existing ≤ 60 s clamped timeout handling (2.7). Accepted source enters the standard build/simulate/lifecycle path with the prompt recorded as provenance (2.5). + +### Plugin_Importer and Module_Listing + +- **Repository import (Requirement 4)**: `plugin_importer.py` validates the URL, then starts a lightweight CodeBuild "fetch" step that clones the repository at the requested revision (default branch when omitted) and syncs the source tree to `plugin-sources/...`. Unreachable repo / missing revision fails before any Plugin_Record is created (4.4). A source-tree scan (presence of a GStreamer plugin build definition: `meson.build`/`configure.ac` with a `gst_plugin` target, or prebuilt `.so`) marks unbuildable imports failed with the finding reported (4.5). Successful fetch creates the Plugin_Record with provenance `{repoUrl, revision, importedBy, importedAt, classification}` (4.2, 15.5) and submits builds (4.3). First successful build prompts Custom_Node_Type declaration (4.6). +- **Module_Listing import (Requirement 6)**: `GET /plugin-modules` returns the parsed module index — fetched from `https://gstreamer.freedesktop.org/modules/`, parsed into `{name, description, repoUrl, classification}` entries, cached in `ModuleIndexCache` with `fetchedAt` and reused for 24 h (6.4). Fetch/parse failure returns an error and the UI offers manual URL entry (6.3). Selecting a module feeds its published repository location into the Requirement 4 path (6.2). +- **Classification (Requirement 15)**: `classify_plugin_set` maps the official plugin-set modules to `good`/`bad`/`ugly` and everything else to `unclassified`. The UI shows the classification as a color-coded risk badge in the module list (15.1) and on the import confirmation view together with the fixed plain-language explanations (15.2, 15.3); the Plugin_Importer stamps it into provenance (15.4, 15.5); the security review screen displays it with the other provenance (15.6). + +### Plugin_Simulator + +A new Step Functions state machine (in a `node-designer-stack.ts` following `test-runner-stack.ts` patterns) runs the existing sandbox image with `HARNESS_MODE=simulate`: + +1. **Guard**: refuse to start when the Plugin_Record version has no successful x86_64 Plugin_Artifact (7.5). +2. **Prepare**: stage the selected Test_Dataset (reusing the existing dataset staging) or the uploaded sample frames; stage the plugin `.so` from the Plugin_Library to the task's plugin scan directory. +3. **RunSandbox** (Fargate, isolated subnet, task role limited to the run's S3 prefix — no Plugin_Library write, no other Use_Case data; Requirement 7.2): the harness renders `multifilesrc ! decode !
+ custom_python_preprocess] --> API[node-catalog API] + API --> PAL[Node_Palette / NodeConfigPanel
no code changes] + CAT --> VAL[Workflow_Validator
generic port checks] + CAT --> CMP[Workflow_Compiler
emlpython + python_handler_path] + CMP --> PKG[Component_Packager
gather predicate widened] + end + PKG -->|"artifact zip: python/{nodeId}/handler.py + requirements.txt"| EDGE + subgraph EDGE[LocalServer edge device] + VEN[vendored workflow_core mirror
byte-identical catalog] --- EXE[WorkflowExecutor] + EXE --> BR[Python_Bridge
appsink/appsrc pair per node] + BR -->|framed stdin/stdout
width/height/format in header| RUN[Python_Runner subprocess] + RUN --> HLP[dda_frames Frame_Helpers
to_array / to_bytes / frame_info / load_image] + RUN --> USR[handler.py
process_frame or handle
cv2 / np pre-bound] + HLP -->|load_image| S3[(local disk / S3)] + end +``` + +Frame flow on the edge for a `process_frame` handler: + +1. The bridge's appsink pulls a sample, reads caps (`width`, `height`, `format`), and sends one framed message (header + raw frame bytes) to the subprocess — unchanged from today. +2. The runner merges `{"frame": {"width": w, "height": h, "format": f}}` into the metadata dict and publishes the same triple to the `dda_frames` per-frame context. +3. If the handler defines `process_frame`, the runner converts the raw bytes to a NumPy uint8 array (handling row padding by slicing each stride-sized row to `width × channels` bytes), invokes `process_frame(frame, metadata)`, verifies the returned array's shape and dtype, and writes the pixels back into a copy of the original byte buffer (preserving padding and total byte length so the appsrc caps stay valid). +4. If the handler defines only `handle`, the existing raw-bytes path runs unchanged (with metadata now carrying the `frame` key). +5. The response message travels back over the existing protocol; the bridge pushes the output buffer with the input buffer's timestamps — unchanged. + +### Design decisions + +- **New node type, not a category flag.** A separate `custom_python_preprocess` descriptor with fixed VideoFrames ports gives users an unambiguous frames-in/frames-out node in the preprocessing palette section, while `custom_python` keeps its flexible per-instance port typing for post-processing. Both share the `emlpython` mapping, so downstream behavior is identical. +- **All runtime additions live in `RUNNER_SOURCE`.** The runner is self-contained by design (the subprocess must not import LocalServer). Embedding the `dda_frames` module source in the runner (registered in `sys.modules` before the handler loads) keeps artifacts unchanged (Requirement 5.1) and works for already-deployed workflows the moment LocalServer updates. +- **`process_frame` output must match input shape/dtype.** The appsrc caps are fixed from the first input sample, so emitting frames of a different size or format would corrupt the pipeline. The runner enforces the constraint and produces a descriptive per-node error instead (Requirement 3.4). +- **Row padding preserved by write-back.** GStreamer buffers may carry row stride padding. `to_array` slices rows by stride; the `process_frame` path writes returned pixels back into a copy of the original buffer bytes, so output byte length always equals input byte length (Requirement 3.2). +- **Pre-imports are best-effort bindings.** `cv2`/`np`/`numpy` are set as module attributes on the handler module *before* `exec_module`, so top-level handler code can use them; import failures leave the binding absent without failing handlers that do not reference it (Requirement 4.3). +- **`load_image` decodes via `cv2.imdecode`** for both local files and S3 objects (uniform behavior, BGR order, honors any format OpenCV can decode). The S3 client is created lazily with boto3 and is injectable for tests. +- **No compiler/serializer/schema changes.** The workflow definition schema stores node `type` as an open string; validation resolves it against the catalog. Adding a descriptor is sufficient. +- **Test sandbox unchanged.** The sandbox inherits whatever behavior `custom_python` has today for the `emlpython` element; the new node type compiles to the identical element chain, so sandbox behavior is identical by construction. Extending sandbox simulation of Custom Python handlers is out of scope. + +## Components and Interfaces + +### 1. Catalog descriptor (portal layer + vendored mirror) + +`workflow_core/catalog/nodes.py` — add after `FORMAT_CONVERT`: + +```python +CUSTOM_PYTHON_PREPROCESS = NodeTypeDescriptor( + type_id="custom_python_preprocess", + category=CATEGORY_PREPROCESSING, + display_name="Custom Python (Frames)", + inputs=[PortDescriptor("in", PORT_TYPE_VIDEO_FRAMES)], + outputs=[PortDescriptor("out", PORT_TYPE_VIDEO_FRAMES)], + parameters=[ + ParameterDescriptor("code", "code", required=True, default=None, + constraints={"min_length": 1}, + description="Python run for every video frame. Define " + "process_frame(frame, metadata) and return the " + "processed frame; frame is a NumPy uint8 array " + "(rows x cols x channels) and cv2/np are " + "pre-imported. Return None to pass the frame " + "through. Helpers: import dda_frames for " + "frame_info(), load_image(path or s3:// URI), " + "to_array(), to_bytes().", + examples=["def process_frame(frame, metadata):\n" + " return cv2.GaussianBlur(frame, (5, 5), 0)"]), + ParameterDescriptor("requirements", "string", required=False, default="", + constraints={}, + description="Extra pip packages the code needs, one per " + "line in requirements.txt form.", + examples=["scikit-image==0.24.0"]), + ], + mappings=_same_on_all_archs( + element_chain=[_element("emlpython", **{"handler-path": "{python_handler_path}"})], + plugin_dependencies=["dda-emlpython"], + ), + hardware_dependent=False, +) +``` + +`CUSTOM_PYTHON_PREPROCESS` is inserted into `NODE_CATALOG` with the other preprocessing types (after `FORMAT_CONVERT`), so `nodes_by_category()` places it in the preprocessing palette section. The `custom_python` descriptor's `code` description/examples are rewritten to state the real contract (Requirement 8): + +``` +description: "Python run for every item passing through the node. Define + process_frame(frame, metadata) to work with video frames as NumPy + arrays (cv2/np pre-imported; import dda_frames for helpers), or + handle(frame_bytes, metadata) -> (frame_bytes, metadata) to work + with raw bytes." +examples: ["def process_frame(frame, metadata):\n return frame", + "def handle(frame_bytes, metadata):\n return frame_bytes, metadata"] +``` + +The vendored copy `src/backend/workflow_engine/vendor/workflow_core/catalog/nodes.py` is updated to stay byte-identical (Requirement 1.6). + +### 2. Component_Packager (`edge-cv-portal/backend/functions/workflow_packaging.py`) + +```python +CUSTOM_PYTHON_NODE_TYPES = ('custom_python', 'custom_python_preprocess') + +def gather_custom_python_nodes(graph) -> List[Dict]: + ... + if node.type in CUSTOM_PYTHON_NODE_TYPES: +``` + +Everything downstream (`build_arch_zip` writing `python/{nodeId}/handler.py` + `requirements.txt`, `build_manifest` emitting `customPythonNodeIds`) is already generic over the gathered list (Requirements 2.4, 2.5). + +### 3. Python_Runner (`src/backend/workflow_engine/python_bridge.py`, `RUNNER_SOURCE`) + +The runner script gains three pieces. The bridge (parent process) and the framed protocol are untouched. + +**(a) `dda_frames` helper module** — embedded as a separate module-level constant `HELPERS_SOURCE` in `python_bridge.py` and prepended into the assembled runner source. Registered before the handler loads: + +```python +_helpers = types.ModuleType("dda_frames") +exec(HELPERS_SOURCE, _helpers.__dict__) +sys.modules["dda_frames"] = _helpers +``` + +Helper API (all functions raise `ValueError` with descriptive messages on bad input): + +```python +FORMAT_CHANNELS = {"RGB": 3, "BGR": 3, "RGBA": 4, "GRAY8": 1} + +def to_array(frame_bytes, width, height, format): + """Raw frame bytes -> NumPy uint8 array (H x W x C; H x W for GRAY8). + Row stride = len(frame_bytes) // height; each row's first + width*channels bytes are taken, tolerating stride padding.""" + +def to_bytes(array): + """NumPy uint8 array -> contiguous raw frame bytes (no padding).""" + +def frame_info(): + """{'width': int, 'height': int, 'format': str} for the frame whose + handler invocation is in progress; None outside an invocation.""" + +def load_image(source, s3_client=None): + """Local path or s3://bucket/key -> BGR uint8 array via cv2.imdecode. + Raises ValueError naming the source on missing file, malformed URI, + fetch failure, or undecodable content. s3_client is injectable for + tests; by default a lazily created boto3 client using the device's + ambient AWS credentials.""" +``` + +`load_image` reads local files with `open(..., 'rb')` (so missing-file errors are uniform) and S3 objects with `get_object`, then decodes with `cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_UNCHANGED)`; a 3-channel result stays BGR, 4-channel input is reduced with `cv2.cvtColor(..., cv2.COLOR_BGRA2BGR)` only when decoded via IMREAD_COLOR — the design uses `cv2.IMREAD_COLOR` for multi-channel sources and `IMREAD_UNCHANGED` yields 2-D arrays for grayscale PNGs, satisfying Requirement 6.1's BGR/2-D contract. PNG round-trip equality (Requirement 6.3) holds for 3-channel BGR arrays written with `cv2.imwrite`. + +**(b) Pre-imports and handler resolution** — in the runner's `main()` after `module_from_spec` and before `exec_module`: + +```python +for name, binding in (("cv2", "cv2"), ("numpy", "np"), ("numpy", "numpy")): + try: + setattr(module, binding, importlib.import_module(name)) + except Exception: + pass # absent binding; handlers not using it are unaffected +``` + +After `exec_module`, entry-point resolution (Requirements 3.6–3.8): + +```python +process_frame = getattr(module, "process_frame", None) +handle = getattr(module, "handle", None) +if not callable(process_frame) and not callable(handle): + error: "handler.py defines neither process_frame(frame, metadata) " + "nor handle(frame_bytes, metadata)" +``` + +`process_frame` wins when both are defined (Requirement 3.7). + +**(c) Frame loop changes** — per message, before invocation: + +```python +metadata = header.get("metadata") or {} +info = {"width": header.get("width"), "height": header.get("height"), + "format": header.get("format")} +metadata["frame"] = info # Requirement 3.9 +dda_frames._set_current(info) # Requirement 5.6 (cleared in finally) +``` + +`process_frame` invocation path: + +- format not in `FORMAT_CHANNELS` or width/height missing → error naming the node's unsupported format (Requirement 3.5); numpy unimportable → error naming the missing library. +- `arr = dda_frames.to_array(frame, w, h, fmt)`; `result = process_frame(arr, metadata)`. +- `result is None` → emit input bytes unchanged (Requirement 3.3). +- ndarray with `result.shape == arr.shape and result.dtype == arr.dtype` → write rows back into `bytearray(frame)` at the original stride, emit (Requirement 3.2). +- anything else → error describing expected vs. actual shape/dtype (Requirement 3.4); the runner reports `status: error` and the bridge raises `CustomPythonNodeError` naming the node, exactly like today's handler exceptions. + +`handle` invocation is unchanged apart from the enriched metadata. Handler-returned metadata continues to flow back in the response header for both contracts. + +### 4. Frontend (no code changes; test coverage only) + +- NodePalette test fixture gains the new descriptor and asserts it renders in the Preprocessing section (Requirement 7.1). +- NodeConfigPanel test asserts the code editor renders for the new type's `code` parameter (Requirement 7.2). +- connectionAcceptance property test domain extended with a fixed-VideoFrames-port node shape (Requirement 7.3). +- inlineChecks test asserts a missing `code` yields a required-parameter marker (Requirement 7.4) — generic behavior, pinned by an example. + +## Data Models + +**Catalog descriptor** — one new `NodeTypeDescriptor` record (shape above); no schema changes to `serializer/schema.py` (node `type` is an open string resolved against the catalog). + +**Framed protocol** — unchanged. Executor → runner header: `{nodeId, width, height, format, metadata, frameSize}`; runner → executor: `{status, metadata, frameSize}` or `{status: "error", error}`. + +**Metadata frame info** (new, runner-side): `metadata["frame"] = {"width": int|None, "height": int|None, "format": str|None}` delivered to every handler invocation. + +**Frame array convention**: NumPy uint8, shape `(height, width, channels)` for RGB/BGR (3) and RGBA (4); shape `(height, width)` for GRAY8. Row stride handling: input bytes of length `L` for height `h` have stride `s = L // h`; row `i` pixels are bytes `[i*s, i*s + width*channels)`. + +**Manifest**: `customPythonNodeIds` (existing key) now lists node ids of both Custom Python node types. + +**Artifact layout** (existing, now also for the new type): `python/{nodeId}/handler.py`, `python/{nodeId}/requirements.txt`. + +## Correctness Properties + +*A property is a characteristic or behavior that should hold true across all valid executions of a system — essentially, a formal statement about what the system should do. Properties serve as the bridge between human-readable specifications and machine-verifiable correctness guarantees.* + +### Property 1: Validator acceptance of preprocessing node connections + +*For any* workflow graph wiring a source node's output port into a `custom_python_preprocess` node's input port, the Workflow_Validator accepts the connection exactly when `are_port_types_compatible(source_type, VideoFrames)` holds under the catalog's declared coercion rules (VideoFrames exactly, and InferenceMeta via the declared coercion), and otherwise reports a port type compatibility error identifying that connection. + +**Validates: Requirements 2.1, 2.2** + +### Property 2: Legacy handle contract is preserved + +*For any* frame bytes and metadata dict, a handler defining only `handle(frame_bytes, metadata)` invoked through the CustomPythonBridge receives the frame bytes unchanged, receives the metadata enriched with the `frame` info key, and its returned `(frame_bytes, metadata)` round-trips back to the bridge caller exactly as under the pre-existing contract. + +**Validates: Requirements 3.6, 3.9** + +### Property 3: Custom Python node gathering and manifest membership + +*For any* workflow graph containing an arbitrary mix of `custom_python` nodes, `custom_python_preprocess` nodes, and other node types with arbitrary node ids, code strings, and requirements strings, `gather_custom_python_nodes` returns exactly the Custom Python nodes of both types with their code and requirements preserved, and `build_manifest`'s `customPythonNodeIds` equals exactly those nodes' ids. + +**Validates: Requirements 2.4, 2.5** + +### Property 4: process_frame contract round trip + +*For any* frame dimensions, supported Pixel_Format, pixel content, and row padding, a `process_frame` handler invoked through the CustomPythonBridge receives a NumPy uint8 array of the shape implied by the caps (rows × columns × channels; 2-D for GRAY8); returning that array unchanged emits output bytes equal to the input bytes (padding included); returning None emits the input bytes unchanged; and returning a deterministic transformation (bitwise inversion) emits bytes whose pixel regions are the transformed pixels while padding bytes and total byte length are preserved. + +**Validates: Requirements 3.1, 3.2, 3.3** + +### Property 5: process_frame contract violations fail the node identifiably + +*For any* frame and any non-compliant `process_frame` outcome — a returned array of different shape, a returned array of different dtype, a non-array non-None return, or an input frame whose Pixel_Format is outside the supported set — the CustomPythonBridge raises `CustomPythonNodeError` carrying the node id and a message describing the mismatch or the unsupported format. + +**Validates: Requirements 3.4, 3.5** + +### Property 6: Frame info delivery + +*For any* frame width, height, and supported Pixel_Format dispatched by the bridge, the handler observes that exact triple both in `metadata["frame"]` and from `dda_frames.frame_info()` during the invocation. + +**Validates: Requirements 3.9, 5.6** + +### Property 7: Frame/array conversion round trip + +*For any* frame dimensions, supported Pixel_Format, and pixel content, `to_bytes(to_array(frame_bytes, width, height, format))` equals the original unpadded frame bytes, and `to_array` of padded frame bytes returns the same array as `to_array` of the unpadded bytes; *for any* unsupported format string or byte string shorter than the dimensions require, `to_array` raises an error describing the format or size problem. + +**Validates: Requirements 5.2, 5.3, 5.4, 5.5** + +### Property 8: Disk image load round trip + +*For any* uint8 BGR image array written losslessly to a PNG file with `cv2.imwrite`, `dda_frames.load_image` of that path returns an array equal to the original. + +**Validates: Requirements 6.1, 6.3** + +### Property 9: S3 image load round trip + +*For any* bucket name, object key, and uint8 BGR image array PNG-encoded and served by an injected fake S3 client, `dda_frames.load_image("s3://bucket/key", s3_client=fake)` requests exactly that bucket and key and returns an array equal to the original. + +**Validates: Requirements 6.2** + +### Property 10: load_image failures identify the source + +*For any* failing source — a non-existent local path, a malformed `s3://` URI, an S3 client raising on fetch, or existing content that does not decode as an image — `dda_frames.load_image` raises an error whose message contains the source. + +**Validates: Requirements 6.4** + +### Property 11: Designer connection acceptance for fixed VideoFrames ports + +*For any* generated pair of nodes where the target is a `custom_python_preprocess`-shaped descriptor (fixed VideoFrames ports, no port type override parameters), the Workflow_Builder connection acceptance function accepts the connection exactly when the source output port type is compatible with VideoFrames under the declared coercion rules (VideoFrames exactly, or InferenceMeta via the declared coercion — mirroring the backend validator per Requirement 2.1) and otherwise rejects it with a reason. + +**Validates: Requirements 7.3** + +### Property 12: Compiled emlpython element per Custom Python preprocessing node + +*For any* valid workflow graph embedding `custom_python_preprocess` nodes with arbitrary node ids, compiling for any architecture yields exactly one `emlpython` element per such node carrying `handler-path` equal to `python/{nodeId}/handler.py` and tagged with that node's id. + +**Validates: Requirements 2.3** + +## Error Handling + +All new failure modes flow through the existing containment path: the runner reports `{status: "error", error:
Python GI, GST_PLUGIN_PATH=build output] + INTRO -->|best effort| RPT[upload {plugin}.so.gstinspect.json
next to the artifact] + end + RPT --> S3[(Portal S3
Plugin_Library custom prefix)] + EVT[EventBridge build result] --> PB[plugin_builds.py
SUCCEEDED path] + PB -->|record gstIntrospection stanza
on the x86_64 artifact entry| DDB[(PluginRecords)] + PB -.->|validates report exists,
size + JSON shape| S3 + + subgraph "Serving" + API[GET /plugins/id/versions/v/gst-properties
plugin_records.py] --> DDB + API --> S3 + API --> MAP[gst_properties.py
pure: filter base-class,
Type_Mapping, required heuristic] + end + + subgraph "Frontend Parameters_Step" + RW[RegistrationWizard] --> PANEL[ParameterScanPanel + useParameterScan] + CW[CreateWizard] --> PANEL + PANEL --> APIC[nodeDesignerApi.getGstProperties] + APIC --> API + PANEL --> MERGE[scan.ts pure merge:
existing rows win, new append] + end +``` + +Flow summary: + +1. `dda-plugin-build` (x86_64 project only) runs `dda-gst-introspect
+ pad template capture] + INTROSPECT -->|report JSON incl. pads| S3[(S3: {plugin}.so.gstinspect.json)] + end + + S3 --> STANZA[plugin_builds.py
build_gst_introspection_stanza
256 KiB cap + parse_report validation
UNCHANGED CODE] + STANZA --> DDB[(Plugin_Record artifact entry
gstIntrospection stanza)] + + subgraph "GET /plugins/{id}/versions/{v}/gst-properties" + ROUTE[plugin_records.py
get_version_gst_properties] + ROUTE --> PARSE[gst_properties.parse_report
+ optional pads field] + PARSE --> PSUG[suggestions_for_element
UNCHANGED] + PARSE --> PORTS[ports_for_element
NEW: Port_Suggestions + Unmapped_Pads] + end + + DDB --> ROUTE + S3 --> ROUTE + + subgraph "Frontend Ports step" + GUIDE[PortGuidancePanel
static portGuidance.ts data
both wizards] + SCAN[PortScanPanel
Registration wizard only] + MERGE[portScan.ts pure helpers
applySuggestions / isUntouchedDefaults] + SCAN --> MERGE + end + + ROUTE -->|portSuggestions, unmappedPads,
padsReason per element| SCAN +``` + +Data flow for a Port_Scan: the panel calls the existing `nodeDesignerApi.getGstProperties` client; the route parses the stored report and returns, per element, the existing `suggestions`/`skipped` plus the new `portSuggestions`/`unmappedPads`/`padsReason`; the panel picks the element with the existing `pickElement` helper (same selection as the Parameter_Scan, Requirement 6.6) and applies the suggestions through the pure merge in `portScan.ts`. + +### What does NOT change + +- `plugin_builds.py`: `gst_report_key`, `build_gst_introspection_stanza`, the 256 KiB cap, and `INTROSPECTION_ARCH` are untouched. The stanza validates the extended report through the same `parse_report` call; an oversized report (now more likely with pads) already yields a failed stanza with the size-cap diagnostic while the build stays succeeded (Requirement 3.3). +- The route's unavailability taxonomy (`no_x86_64_build`, `not_captured`, `introspection_failed`) and the existing per-element `suggestions`/`skipped` response fields (Requirement 4.6). +- `dda-plugin-build`'s `run_introspection` shell step: it remains a dumb pipe for the report document. +- The Parameters step, `ParameterScanPanel`, `scan.ts` merge helpers, and all parameter wire shapes. + +## Components and Interfaces + +### 1. Capture: `dda-gst-introspect` (plugin-build-images/) + +`describe_factory` is extended to capture static pad templates from the **loaded factory** (pad templates are class-level metadata available from `factory.get_static_pad_templates()` — instantiation is not required, so elements with an `instantiationError` can still report pads when the factory loaded). + +New pure helpers (module top level, importable without GI — the `gi` imports stay inside `scan()`): + +```python +MAX_CAPS_LEN = 4096 + +def truncate_caps(caps): + """(caps_str, truncated_flag): first MAX_CAPS_LEN chars and whether + truncation occurred (Requirement 3.4).""" + if len(caps) > MAX_CAPS_LEN: + return caps[:MAX_CAPS_LEN], True + return caps, False + +def describe_pad_templates(factory, Gst): + """(pads, pads_error): the pad list and None, or [] and a diagnostic + when reading any template of this factory fails (Requirement 3.2).""" +``` + +`describe_pad_templates` maps each `Gst.StaticPadTemplate` via the enum tables `Gst.PadDirection.SINK/SRC -> 'sink'/'src'` and `Gst.PadPresence.ALWAYS/SOMETIMES/REQUEST -> 'always'/'sometimes'/'request'`; the caps string comes from `template.get_caps().to_string()`. Any exception while reading an element's templates (including an unmappable direction/presence value) degrades that one element to `pads: []` with a `padsError` diagnostic, leaving its property data and the report status untouched (Requirement 3.2). An element that genuinely declares no templates records `pads: []` with `padsError: null` (Requirement 3.5). + +Both the load-failure and instantiation-failure early returns in `describe_factory` also emit `pads: []` with a `padsError` describing the failure (pad reading was not possible), keeping every element in a new report explicitly pad-captured. + +### 2. Parsing/serialization: `backend/functions/gst_properties.py` + +New constants and dataclass: + +```python +PAD_DIRECTION_SINK, PAD_DIRECTION_SRC = 'sink', 'src' +VALID_PAD_DIRECTIONS = (PAD_DIRECTION_SINK, PAD_DIRECTION_SRC) +PAD_PRESENCE_ALWAYS = 'always' +VALID_PAD_PRESENCES = ('always', 'sometimes', 'request') +MAX_CAPS_LEN = 4096 + +@dataclass(frozen=True) +class PadTemplate: + """One static Pad_Template captured from the element factory.""" + name: str # name template, e.g. 'sink', 'src', 'src_%u' + direction: str # 'sink' | 'src' + presence: str # 'always' | 'sometimes' | 'request' + caps: str # caps string, at most MAX_CAPS_LEN chars + caps_truncated: bool # True when capture truncated the caps (3.4) +``` + +`ReportElement` gains two optional fields with legacy-compatible defaults: + +```python +@dataclass(frozen=True) +class ReportElement: + factory: str + element_gtype: str + instantiation_error: Optional[str] = None + properties: List[GstProperty] = field(default_factory=list) + pads: Optional[List[PadTemplate]] = None # None = not captured (legacy) + pads_error: Optional[str] = None # meaningful only when pads is not None +``` + +**Domain invariant**: `pads_error` is non-None only when `pads == []` (a per-element read failure, Requirement 3.2). Generators and serialization respect this. + +Parsing (`_parse_element`): +- `pads` key **absent** → `pads=None, pads_error=None` (legacy report, Requirement 4.2). Any stray `padsError` without `pads` is ignored. +- `pads` key **present** → must be a list; each entry must be an object with all five fields present and correctly typed: `name` str, `direction` in `VALID_PAD_DIRECTIONS`, `presence` in `VALID_PAD_PRESENCES`, `caps` str of length ≤ `MAX_CAPS_LEN`, `capsTruncated` bool. `padsError` parses as optional str. Any violation raises `ReportError` — the route already maps that to `introspection_failed` (Requirement 4.4). The caps length bound makes the truncation contract enforceable in the pure module. + +Serialization (`_serialize_element`): when `pads is None` both keys are omitted (byte-identical output to today for legacy-shaped reports); otherwise `pads` (each pad as `{name, direction, presence, caps, capsTruncated}`) and `padsError` are emitted explicitly. This preserves the round-trip identity `parse_report(serialize_report(r)) == r` over the whole (extended) valid domain (Requirement 4.3). + +### 3. Derivation: `ports_for_element` (new, pure, in `gst_properties.py`) + +```python +PORT_TYPE_VIDEO_FRAMES = 'VideoFrames' +CONFIDENT_CAPS_PREFIX = 'video/x-raw' # exact, case-sensitive (5.2) + +PADS_REASON_NOT_CAPTURED = 'pads_not_captured' # report predates pad capture (4.7) +PADS_REASON_NO_TEMPLATES = 'no_pad_templates' # element declares none (4.8) +PADS_REASON_READ_FAILED = 'pads_read_failed' # per-element capture failure (3.2) + +def ports_for_element(element: ReportElement) -> Dict[str, Any]: + """{'portSuggestions': [...], 'unmappedPads': [...], + 'padsReason': str | None, 'padsMessage': str | None}""" +``` + +Reason classification (mutually exclusive, Requirements 4.7, 4.8): + +| Element state | `padsReason` | `padsMessage` | +|---|---|---| +| `pads is None` (legacy) | `pads_not_captured` | `None` | +| `pads == []`, `pads_error` set | `pads_read_failed` | the diagnostic | +| `pads == []`, no error | `no_pad_templates` | `None` | +| `pads` non-empty | `None` | `None` | + +Derivation walks `element.pads` **in report order** (Requirement 5.1); each pad lands in exactly one output list: + +1. `presence != 'always'` → **Unmapped_Pad** `{name, direction, presence, caveat}` with the caveat that sometimes/request pads are created at runtime and do not correspond to fixed declared Ports (Requirement 5.4). +2. `presence == 'always'` but `name.strip()` is empty (the existing Ports_Step validation rule: non-empty Port name) → **Unmapped_Pad** with the caveat that the name template is not a valid Port name (Requirement 5.6). +3. Otherwise → **Port_Suggestion**: + - `direction`: `sink` → `'input'`, `src` → `'output'` (Requirement 5.1) + - `name`: the pad's name template verbatim + - `portType`: always `'VideoFrames'` (the only caps-derivable catalog type, Requirement 5.5) + - `confident`: `caps.startswith('video/x-raw')` — truncated caps included, since truncation preserves the prefix (Requirement 5.2) + - `reason`: confident → states the caps begin with `video/x-raw`; unconfirmed → states that InferenceMeta and EventSignal are DDA semantic concepts GStreamer caps cannot express and the Port_Type needs user confirmation (Requirement 5.3) + - `caps`, `capsTruncated`: carried through for display (Requirement 6.4) + +The function is pure over immutable inputs — determinism (Requirement 5.7) is by construction and verified by property test. + +### 4. Route: `get_version_gst_properties` (plugin_records.py) + +The per-element response entry is extended in place; existing fields keep their exact names, structure, and values (Requirement 4.6): + +```python +derived = suggestions_for_element(element) # unchanged +ports = ports_for_element(element) # new +elements.append({ + 'factory': element.factory, + 'suggestions': derived['suggestions'], # unchanged (4.6) + 'skipped': derived['skipped'], # unchanged (4.6) + 'portSuggestions': ports['portSuggestions'], # new (4.5) + 'unmappedPads': ports['unmappedPads'], # new (4.5) + 'padsReason': ports['padsReason'], # new (4.7, 4.8) + 'padsMessage': ports['padsMessage'], # new (3.2 surfacing) +}) +``` + +No other route logic changes: malformed pad data is caught inside the existing `parse_report` call and flows through the existing `ReportError → introspection_failed` mapping (Requirement 4.4). + +### 5. Frontend: static guidance (`portGuidance.ts` + `PortGuidancePanel.tsx`) + +`portGuidance.ts` — pure data and one pure function, no imports beyond `types.ts`: + +```typescript +/** Static Port_Guidance content (1.1–1.3): definition, connection rule, + * input/output distinction, per-type description + usage example. */ +export const PORT_DEFINITION: string; +export const CONNECTION_RULE: string; +export const INPUT_OUTPUT_DISTINCTION: string; +export const PORT_TYPE_GUIDANCE: Record
PortalAdmin-only item] --> UM[UserManager page
/admin/user-manager] + UM --> LIST[Account table + filter] + UM --> ACT[Create / Password / Forgot / Role /
Disable-Enable / Delete modals] + UM --> SYNC[Edge sync panel
per-device status] + end + + subgraph Portal backend + API[API Gateway + jwt_authorizer] --> UA[user_admin.py Lambda
PortalAdmin gate] + UA --> COG[(Cognito User_Pool
custom:role)] + UA --> AUD[(Audit log table
two-phase entries)] + UA --> CRED[(dda-portal-edge-credentials
PBKDF2 verifiers)] + UA --> SES[SES: temp password email] + UA --> SS[(dda-portal-account-sync
per-device sync state)] + SCHED[EventBridge rate 5 min] --> SYNCFN[account_sync.py Lambda] + SYNCFN --> SS + SYNCFN --> SHADOW + ACKQ[SQS ack queue] --> ACKFN[account_sync.py ack ingest] + ACKFN --> SS + end + + subgraph AWS IoT + SHADOW[(Named shadow
dda-user-accounts)] + RULE[Topic rule on
shadow update/documents] --> ACKQ + SHADOW --> RULE + end + + subgraph Edge device / LocalServer + AGENT[user_accounts_sync agent
IoTShadowAccessor + MQTT delta] --> CACHE[(Local_Credential_Cache
/aws_dda/local_credential_cache.json)] + AGENT -->|reported ack| SHADOW + SHADOW -->|desired delta| AGENT + CFG[Greengrass config
LocalLoginEnabled] --> AUTHDEP[authorize_request dependency
decision matrix] + LOGIN[POST /local-auth/login] --> CACHE + LOGIN --> TOK[Local_Session_Token
HMAC, 12 h] + AUTHDEP --> TOK + AUTHDEP --> ETA[Existing validate_token
authorization_settings.json] + WEBUI[react-webapp
login screen] --> LOGIN + end +``` + +### Portal → edge sync sequence + +```mermaid +sequenceDiagram + participant A as Portal_Administrator + participant UA as user_admin.py + participant SS as Sync state table + participant SF as account_sync.py + participant SH as Shadow dda-user-accounts + participant AG as Edge sync agent + participant CC as Local_Credential_Cache + + A->>UA: POST /admin/edge-sync/devices/{id} (selected accounts) + UA->>SS: stage pending account set + new syncId + UA->>SF: invoke sync attempt (also runs every 5 min) + SF->>SH: update desired {syncId, accounts} + SF->>SS: status=in_progress, attemptAt=now + SH-->>AG: delta notification (MQTT) + AG->>CC: atomic replace of cache file + AG->>SH: update reported {ackSyncId, appliedAt, accountCount} + SH-->>SF: topic rule -> SQS -> ack ingest + SF->>SS: status=success, lastSyncAt (ack matches syncId) + Note over SF,SS: No ack within 60 s of attempt:
status=failed "device unreachable",
pending retained, retried ≤ every 5 min (7.6, 7.7, 7.9) +``` + +### Edge request authorization decision matrix + +Evaluated per request by the new `authorize_request` FastAPI dependency (replacing the startup-frozen `Depends(validate_token)` wiring in `get_api_router()`): + +| Local_Login_Configuration | Existing_Token_Auth (settings file present) | Decision | +|---|---|---| +| disabled | not configured | **Allow** — open access, exactly today's behavior (9.1, 9.3) | +| disabled | configured | Require valid bearer token via existing `validate_token`; else 401 (9.4, 10.4) | +| enabled | not configured | Require valid Local_Session_Token; else 401 (8.5, 8.9) | +| enabled | configured | Allow if **either** a valid bearer token **or** a valid Local_Session_Token is presented; else 401 (10.1, 10.2, 10.5) | + +Exemptions in every configuration: `POST /local-auth/login`, `GET /local-auth/status`, and static SPA assets. When local login is disabled, the login endpoint rejects with a "local login disabled" error and never issues a token (9.5). The dependency never reads or writes `authorization_settings.json` beyond the existing presence check (10.3). + +## Components and Interfaces + +### 1. Portal frontend + +**Settings dropdown (`Layout.tsx`)** — add a `user-manager` item to the existing top-nav settings `ButtonDropdown`. Unlike the current `settings` item (disabled for non-admins), the User Manager item is *omitted entirely* for non-PortalAdmin users (1.1, 1.2): the items array is built conditionally on `user?.role === 'PortalAdmin'`. Selecting it navigates to `/admin/user-manager` (1.3). + +**Route guard (`App.tsx` + `UserManager.tsx`)** — the route is registered inside the existing authenticated layout (unauthenticated visitors hit the existing redirect to `/login`, 1.6). The page component itself renders an access-denied `Alert` and nothing else when `user?.role !== 'PortalAdmin'` (1.4), mirroring the `BedrockConfigurationSettings` "Portal Admin access required" pattern. + +**`UserManager.tsx` page** — Cloudscape `Table` + `TextFilter` + action modals: + +- Loads accounts via `GET /api/v1/admin/users` (paginated passthrough of Cognito `list_users`). Columns: username, email, Portal_Role, Cognito status (`CONFIRMED`, `FORCE_CHANGE_PASSWORD`, …), enabled/disabled, edge-login-capable (verifier present) (2.1). +- Filtering is a pure exported function `filterAccounts(accounts, term)`: case-insensitive substring match on username or email; empty result renders the table's empty state without an error (2.2, 2.3); clearing the term restores the full list (2.4). +- Load failure replaces the table body with an error `Alert` — never a partial/stale list (2.5): the accounts state is cleared before each fetch and only set on success. +- **Password modal**: password + confirm fields, a required `RadioGroup` (permanent / temporary — no default selection, submit disabled until chosen, 3.2), client-side policy pre-check mirroring the pool policy, server errors surfaced verbatim including the violated policy rule (3.3). Success flashbar names the account (3.4). +- **Forgot-password action**: confirmation modal; on success shows "temporary password sent to the account's registered email" without ever receiving the password value from the API (4.3); surfaces no-verified-email and delivery errors (4.4, 4.5). +- **Role modal**: `Select` restricted to the five defined roles with the current role preselected (5.2); on success shows confirmation and re-fetches the list (5.7); rejection reasons (including the last-PortalAdmin guard, 5.3) shown in the modal. +- **Create User modal** (`UserManagerModals.tsx`): username, email, and role fields, the role a `Select` restricted to the five defined roles (12.2); client-side pre-checks mirror the server validation (non-empty fields, email shape per 12.6) with submit disabled until valid; server rejection reasons (duplicate username 12.5, invalid email 12.6, missing field 12.7) surfaced verbatim in the modal. Success flashbar states that an invitation with a temporary password was sent to the account's email — the password value is never present in the API response — and the list re-fetches to include the new account (12.10). +- **Disable/Enable confirm modal**: explicit confirmation naming the affected account by username before submission (13.1); on success shows a confirmation identifying the account and re-fetches the list (13.8); an already-in-requested-state response simply re-fetches to show the current state (13.6); rejection reasons (last-PortalAdmin guard for disable, 5.3/13.9) and failures (13.7) shown in the modal. +- **Delete confirm modal**: explicit confirmation naming the affected account by username (14.1); cancel/dismiss submits nothing (14.9); on success shows a confirmation identifying the deleted account and re-fetches the list without it (14.7); rejection reasons (last-PortalAdmin guard, 14.3), not-found (14.11), and partial verifier-cleanup errors (14.10) surfaced in the modal or flashbar. +- **Edge sync panel**: device list from `GET /api/v1/admin/edge-sync/devices` showing last sync status + timestamp per device (7.4); a sync action with account multi-select posts to `POST /api/v1/admin/edge-sync/devices/{deviceId}` (7.1). + +### 2. Portal backend — `user_admin.py` Lambda + +New function `edge-cv-portal/backend/functions/user_admin.py`, shared layer reused, routed under `/api/v1/admin/*` behind the existing `jwt_authorizer` (requests without a valid JWT are rejected by the authorizer before the Lambda runs, 1.7). Every handler first asserts `get_user_from_event(event)['role'] == 'PortalAdmin'` and returns 403 otherwise (1.5). + +| Endpoint | Operation | +|---|---| +| `GET /api/v1/admin/users` | Cognito `list_users` (paginate all), join with the edge-credentials table for the `edgeCapable` flag. Returns username, email, `email_verified`, `custom:role` (default `Viewer`), `UserStatus`, `Enabled` | +| `POST /api/v1/admin/users` | Body `{username, email, role}`. Validate (all fields present and non-empty 12.7, email shape 12.6, role in the five defined values 12.8) → audit-pending → `admin_create_user` with `custom:role`, `email`, `email_verified=true`, default email invitation (D12) → audit-final. `UsernameExistsException` → 409 "username already exists" (12.5) | +| `POST /api/v1/admin/users/{username}/password` | Body `{password, permanent: bool}`. Audit-pending → `admin_set_user_password(Permanent=permanent)` → verifier capture → audit-final | +| `POST /api/v1/admin/users/{username}/forgot-password` | Verified-email check → generate temp password → audit-pending → `admin_set_user_password(Permanent=False)` → SES email → verifier capture → audit-final | +| `PUT /api/v1/admin/users/{username}/role` | Body `{role}`. Validate against the five roles → last-admin guard → audit-pending → `admin_update_user_attributes(custom:role)` → audit-final (records previous and new role, 5.4) | +| `POST /api/v1/admin/users/{username}/disable` | `admin_get_user` → already disabled: no-op 200 with the current state, no mutation (13.6) → last-admin guard (D14, 13.9) → audit-pending → `admin_disable_user` (13.2) → mark sync staging pending with `enabled: false` (7.2, 7.8) → audit-final | +| `POST /api/v1/admin/users/{username}/enable` | `admin_get_user` → already enabled: no-op 200, no mutation (13.6) → audit-pending → `admin_enable_user` (13.3) → mark sync staging pending with `enabled: true` (7.2) → audit-final | +| `DELETE /api/v1/admin/users/{username}` | `admin_get_user` (captures username/email/role for the audit entry, 14.8; `UserNotFoundException` → 404, 14.11) → last-admin guard (D14, 14.3/14.4) → audit-pending → `admin_delete_user` (14.2) → delete the `dda-portal-edge-credentials` record (14.5, D13) → mark sync staging pending with `enabled: false, deleted: true` (7.8) → audit-final | +| `GET /api/v1/admin/edge-sync/devices` | Devices table join with sync-state table: per-device `lastSyncStatus`, `lastSyncAt`, `pendingChanges` | +| `POST /api/v1/admin/edge-sync/devices/{deviceId}` | Body `{usernames: [...]}`. Stages the selected account records + fresh `syncId` in the sync-state table and invokes the sync Lambda for an immediate attempt | + +**Audit protocol (strict, D10)** — a new shared-layer helper `record_audit_event_strict(...)` that `put_item`s and **raises** on failure, plus `finalize_audit_event(event_id, result, details)`. Flow per mutating action: + +1. `record_audit_event_strict(acting_user, action, affected_account, result='pending')` — on failure: return 500 "action not applied", Cognito untouched (6.4, 6.5). +2. Perform the Cognito operation (and SES send for forgot-password). +3. `finalize_audit_event(..., result='success' | 'failure', details)` — details carry previous/new role for role changes (5.4) and rejection reasons for rejected attempts (5.5). Details **never** include passwords, hashes, or temporary password values (6.3); the helper drops keys matching a denylist (`password`, `verifier`, `hash`, `temp*`) defensively. +4. Self-service actions record the affected account as the acting user (6.2) — acting identity always comes from the validated JWT claims, which for self-service *is* the account holder. + +**Last-PortalAdmin guard (5.3, 13.9, 14.3)** — Cognito `list_users` cannot filter on custom attributes, so the guard paginates the pool and counts users with `custom:role == 'PortalAdmin'` and `Enabled == true`. If the action would reduce that count to zero (role change away from PortalAdmin, disable, or **delete**, targeting the last such account), reject with 409 and an explanatory reason, and record the rejected attempt in the audit log (5.5, 13.9, 14.4). The same shared predicate serves all three action types (D14). + +**Account creation validation (12.6–12.8)** — a pure function `validate_create_request(body)` gates the create endpoint: all three fields present and non-empty (rejections name the missing field, 12.7), email consisting of a non-empty local part, `@`, and a non-empty domain containing at least one dot (12.6), and role in the five defined Portal_Role values (12.8). Only a payload passing all checks reaches `admin_create_user` — a rejection performs no User_Pool call, so no account or partial record can exist (12.5 duplicate rejection is likewise atomic on the Cognito side, 12.9). Creation records an `account_create` audit entry carrying the created account's username, email, and role (12.11). No verifier is captured at creation — the invitation password is never held by the portal (D12); the account becomes edge-login-capable when an administrator later sets its password (D3). + +**Disable/enable state transitions (13.2, 13.3, 13.6)** — both endpoints read the account's current `Enabled` state first: when the account is already in the requested state, the handler returns success with the current state and performs no Cognito mutation, no audit-pending write, and no sync staging (13.6). Otherwise the standard audit-before-effect flow runs, and the enabled/disabled flag — a synchronized account attribute — triggers `_mark_account_change_pending` so every configured device's staged set carries the new state (7.2; disable additionally satisfies 7.8's mark-as-disabled-on-next-sync). + +**Deletion flow (D13)** — `admin_get_user` first captures the username, email, and role for the audit entry (14.8) and maps a missing account to 404 without any mutation (14.11); then the guard, audit-pending, `admin_delete_user`, verifier-record delete, and sync staging (`enabled: false, deleted: true`, 7.8) run in that order. A Cognito failure aborts before the verifier record is touched (14.6); a verifier-delete failure after the Cognito delete finalizes the audit entry with a partial-cleanup detail, retains the verifier record for a subsequent attempt, and returns an error stating the account was deleted but its verifier record was not removed (14.10). + +**Temporary password generator** — pure function `generate_temp_password(length=16)`: guarantees ≥1 character from each required class (lower, upper, digit, symbol) and length ≥ the pool minimum (12), assembled with `secrets.choice` and shuffled with `secrets.SystemRandom().shuffle` (4.1). + +**Verifier capture (D3, D4)** — pure function `make_verifier(password)` → `{algorithm: 'pbkdf2-sha256', iterations: 210000, salt: b64, hash: b64}`; stored in `dda-portal-edge-credentials` keyed by normalized username with `updatedAt`. Written after every successful password set (permanent or temporary). A verifier update marks every configured device's sync state as having pending changes (7.2). + +**Cognito error mapping** — `InvalidPasswordException` → 400 with the policy message passed through (3.3); other exceptions → 502 "password change failed" (3.5); the account is untouched in both cases because `admin_set_user_password` is atomic. Forgot-password on an account whose `email_verified != 'true'` → 400 before anything is generated (4.4); SES failure after the temp password was applied is compensated by restoring nothing (the account is now in FORCE_CHANGE_PASSWORD with an undelivered password) — to honor 4.5 ("preserve existing credentials unchanged"), the SES send is performed **before** `admin_set_user_password`: generate → email → set-password. If the email send fails, no credential was modified; if the set-password fails after a successful send, the emailed password is inert (it never became valid) and the action reports failure. Requirement 4.6 (consumed temporary password rejected) is native Cognito `FORCE_CHANGE_PASSWORD` semantics. + +### 3. Account_Sync_Service — portal side (`account_sync.py` Lambda) + +One Lambda with three entry paths (mirroring `camera_sync.py`'s style): + +- **Sync attempt** (invoked directly by `user_admin.py` and by the EventBridge `rate(5 minutes)` schedule for every device with pending changes, 7.7): builds the device's full desired account document from the staged account set, writes the `dda-user-accounts` shadow `desired` state via `iot_data_client(...).update_thing_shadow`, and stamps the sync-state row `in_progress` with `attemptAt`. A shadow-write failure marks the attempt failed with the error reason; pending changes are retained (7.6). +- **Ack ingest** (SQS from the topic rule on `$aws/things/+/shadow/name/dda-user-accounts/update/documents`, same partial-batch-failure pattern as camera sync): a reported `ackSyncId` equal to the device's current `syncId` marks the row `success` with `lastSyncAt = appliedAt` and clears `pendingChanges` (7.4); a reported `error` marks `failed` with the device's reason. Stale acks (unknown `syncId`) are discarded. +- **Timeout sweep** (piggybacked on the 5-minute schedule): any `in_progress` row whose `attemptAt` is older than 60 s without a matching ack is marked `failed` with reason `device unreachable` (7.9) — pending changes retained, so the next scheduled attempt retries (7.6, 7.7). + +The desired document always carries the **complete selected account set** (not a diff), so duplicate or out-of-order delivery is idempotent, and a sync with zero changes is simply a desired write whose content equals the previous one — acked and reported successful (7.5). Account disable/delete in the portal marks the account `enabled: false` in the staged set (delete also flags `deleted: true` for eventual cache pruning), never removing the record silently (7.8). Payloads contain only `{username, email, role, enabled, verifier?}` — never plaintext passwords (7.3, guaranteed structurally: the sync path has no access to plaintext). The builder validates the rendered document against the 8 KB shadow size limit and fails the sync with an explicit reason if exceeded. + +**New infrastructure (ComputeStack additions)**: `dda-portal-edge-credentials` and `dda-portal-account-sync` DynamoDB tables, the ack SQS queue + DLQ, the IoT topic rule, the EventBridge schedule, SES send permission + sender-address parameter, and `cognito-idp:AdminSetUserPassword/AdminUpdateUserAttributes/ListUsers/AdminGetUser` grants scoped to the pool for `user_admin.py` only. The account life-cycle actions (Requirements 12–14) extend the same `user_admin.py`-scoped policy with `cognito-idp:AdminCreateUser/AdminEnableUser/AdminDisableUser/AdminDeleteUser` — no new resources are required. + +### 4. Edge sync agent — `src/backend/user_accounts_sync/agent.py` + +Mirrors `camera_sync/agent.py`: started from `server_setup.py` on a daemon thread; on start it `GetThingShadow`s `dda-user-accounts` (thing name from `AWS_IOT_THING_NAME`) to catch up on any desired state that arrived while offline, then subscribes to `$aws/things/{thing}/shadow/name/dda-user-accounts/update/delta` through the existing MQTT `SubscriptionHandler` pattern. On receiving a desired document it: + +1. Validates the document shape (pure function `parse_sync_document`). +2. Atomically replaces the Local_Credential_Cache file (write temp file `0600`, `os.replace`) — full-set replacement makes application idempotent. +3. Writes `reported: {ackSyncId, appliedAt, accountCount}`; on validation failure writes `reported: {ackSyncId, error: reason}` so the portal records the failure rather than timing out. + +Offline behavior costs nothing extra: the shadow retains `desired`, and the startup `GetThingShadow` is the catch-up path. Agent crashes are logged and never take down the rest of LocalServer. + +### 5. Edge local auth — `src/backend/local_auth/` + +**`credential_cache.py`** — load/parse the cache file with `verify_credentials(username, password) -> Account | None`: constant-shape verification that computes PBKDF2 against the stored (or, for unknown usernames, a dummy) verifier so the 401 response and timing are identical whether or not the username exists (8.3); returns the account only when the verifier matches **and** `enabled` is true. Purely local — no network (8.4). Missing/empty cache ⇒ every login fails with the same 401 plus a logged "no synchronized accounts available" diagnostic (11.3). + +**`session_tokens.py`** — pure token functions over an injected clock: +- `issue_token(secret, username, role, now) -> str`: payload `{sub, role, iat: now, exp: now + 12*3600, jti}`, signature `HMAC-SHA256(secret, payload_b64)` (8.2). +- `validate_token(secret, token, now, cache) -> Account | AuthError`: reject on malformed structure, bad signature (`hmac.compare_digest`), `exp <= now` (8.7); then re-check the cache — account absent or disabled ⇒ reject (8.5, 8.6, D6). +- Secret management: `get_or_create_secret()` reads `/aws_dda/local_session_secret`, creating 32 random bytes (`secrets.token_bytes`) with mode `0600` on first use. + +**`lockout.py`** — `LockoutTracker` (in-memory, injected clock): `record_failure(username)`, `record_success(username)` (resets the counter), `is_locked(username, now)`. 5 consecutive failures ⇒ locked for 15 minutes; while locked, every attempt is rejected regardless of credentials and does not extend the window (8.10). + +**`config.py`** — `LocalLoginConfig` singleton: reads `LocalLoginEnabled` via IPC `GetConfiguration` at startup (11.1) and re-polls on a 30 s background timer (11.2); any read error, missing key, or non-boolean value ⇒ disabled (11.4), logged once per transition. + +**`endpoints/local_auth.py`** — on the unauthenticated router: +- `GET /local-auth/status` → `{localLoginEnabled: bool}` (drives the SPA, D8). +- `POST /local-auth/login` `{username, password}` → when disabled: 403 `{error: "local login is disabled"}`, never issues a token (9.5). When enabled: lockout check → `verify_credentials` → on success `record_success` + `issue_token` → `{token, expiresAt, role, username}`; on failure `record_failure` + uniform 401 (8.2, 8.3, 8.10). + +**`utils/auth.py` extension — `authorize_request`** dependency implementing the decision matrix table above. It extracts the bearer credential once; a well-formed Local_Session_Token (two-segment base64url) is validated locally first, otherwise/on failure the existing remote `validate_token` path runs when Existing_Token_Auth is configured. `get_api_router()` switches from the frozen `Depends(validate_token)` to `Depends(authorize_request)` unconditionally — the dependency itself decides per request, which is what makes runtime configuration changes (11.2) effective without process restart. The download-file query-param token path gains the same either/or acceptance. + +### 6. Edge web UI — `src/react-webapp` + +A `LoginGate` component wraps the app: it fetches `/local-auth/status`; when enabled and no unexpired token is in `sessionStorage`, it renders the login screen (username/password form posting to `/local-auth/login`) instead of the app (8.1, 8.8). On success the token is stored and attached as `Authorization: Bearer` on all API calls. When disabled, the gate renders the app directly — no login screen, no prompt (9.2). API 401 responses clear the stored token and return to the login screen (token expiry mid-session). + +## Data Models + +### Cognito account (existing, read/written via admin APIs) + +`username`, `email`, `email_verified`, `custom:role` (one of the five Portal_Role values, default `Viewer`), `UserStatus` (`CONFIRMED` | `FORCE_CHANGE_PASSWORD` | ...), `Enabled`. + +### `dda-portal-edge-credentials` (DynamoDB, new) + +| Attribute | Type | Notes | +|---|---|---| +| `username` (PK) | S | normalized (lowercase) | +| `verifier` | M | `{algorithm: 'pbkdf2-sha256', iterations: N, salt: b64, hash: b64}` | +| `updatedAt` | N | epoch ms | + +Never contains plaintext; written only by `user_admin.py` password flows. + +### `dda-portal-account-sync` (DynamoDB, new) + +| Attribute | Type | Notes | +|---|---|---| +| `device_id` (PK) | S | IoT thing name | +| `syncId` | S | UUID of the latest staged sync | +| `accounts` | M | staged full account set `{username: {email, role, enabled, deleted?, verifier?}}` | +| `status` | S | `pending` \| `in_progress` \| `success` \| `failed` | +| `failureReason` | S | e.g. `device unreachable` | +| `attemptAt` / `lastSyncAt` | N | epoch ms | +| `pendingChanges` | BOOL | true from staging until a matching ack | + +### Shadow `dda-user-accounts` document + +```jsonc +{ + "state": { + "desired": { + "syncId": "3f2a...", + "version": 1, // document schema version + "accounts": { + "operator1": { + "email": "op1@example.com", + "role": "Operator", + "enabled": true, + "verifier": { "algorithm": "pbkdf2-sha256", "iterations": 210000, + "salt": "b64...", "hash": "b64..." } + }, + "olduser": { "email": "x@y.z", "role": "Viewer", "enabled": false } + } + }, + "reported": { + "ackSyncId": "3f2a...", "appliedAt": 1700000000000, "accountCount": 2 + // or: "ackSyncId": "3f2a...", "error": "reason" + } + } +} +``` + +### Local_Credential_Cache — `/aws_dda/local_credential_cache.json` (mode 0600) + +```jsonc +{ + "version": 1, + "syncId": "3f2a...", + "appliedAt": 1700000000000, + "accounts": { + "operator1": { "email": "...", "role": "Operator", "enabled": true, + "verifier": { "algorithm": "pbkdf2-sha256", "iterations": 210000, + "salt": "b64...", "hash": "b64..." } } + } +} +``` + +### Local_Session_Token + +`base64url(json({sub, role, iat, exp, jti})) + "." + base64url(hmac_sha256(secret, payload_b64))` — `exp = iat + 43200` (12 h). Secret: `/aws_dda/local_session_secret`, 32 bytes, mode 0600. + +### Component configuration (recipe `DefaultConfiguration` addition) + +```yaml +LocalLoginEnabled: "false" # per-device override via deployment configuration merge +``` + +### Audit log entry (existing table, new action types) + +`action ∈ {password_change, forgot_password, role_change, account_create, account_disable, account_enable, account_delete}`, `resource_type = 'user_account'`, `resource_id =
Workflow_Builder canvas] + APIGW[API Gateway] + WFAPI[workflows.py Lambda
Workflow_Store API] + VAL[Workflow_Validator +
Workflow_Compiler
Lambda layer] + GEN[workflow_generator.py Lambda
Bedrock invocation] + PKG[workflow_packaging.py Lambda
Component_Packager] + TEST[Step Functions +
Fargate test sandbox
Workflow_Test_Runner] + DDB[(DynamoDB
Workflows, WorkflowVersions,
TestDatasets, TestRuns)] + S3P[(S3
definitions, compiled docs,
test datasets, results)] + end + subgraph UseCase Account + GGREG[Greengrass Component Registry] + GGDEP[Greengrass Deployments] + S3U[(UseCase S3
component artifacts)] + end + subgraph Edge Device + GG[Greengrass Nucleus] + WFC[Workflow_Component
artifacts on disk] + LS[LocalServer
workflow executor +
existing pipeline path] + TRITON[Embedded Triton] + end + BR[Amazon Bedrock] + + FE --> APIGW --> WFAPI + WFAPI --> VAL + WFAPI --> DDB + WFAPI --> S3P + APIGW --> GEN --> BR + APIGW --> PKG + PKG -->|STS AssumeRole| GGREG + PKG --> S3U + WFAPI -->|existing Deployment_Service| GGDEP + WFAPI --> TEST + TEST --> S3P + GGDEP --> GG --> WFC + LS -->|discovers| WFC + LS --> TRITON +``` + +### Portal Backend Flow + +New Lambda functions follow the established pattern (`rbac_middleware` decorator, per-use-case STS AssumeRole, DynamoDB + S3): + +- `workflows.py` — CRUD, versioning, duplicate, list, delete-with-deployment-check (Workflow_Store) +- `workflow_validation.py` — validate endpoint; imports the shared `workflow_core` layer +- `workflow_packaging.py` — compile + package + register Greengrass component (Component_Packager) +- `workflow_generator.py` — Bedrock chat sessions (Workflow_Generator) +- `workflow_testing.py` — test dataset upload/list, test run start/status/results (fronts the Workflow_Test_Runner state machine) + +A shared Python package `workflow_core` (deployed as a Lambda layer under `edge-cv-portal/backend/layers/`, also vendored into LocalServer and the test sandbox image) contains the node catalog, serializer, validator, and compiler. This guarantees the portal, the test runner, and the edge agree on schema and semantics. + +### Edge Execution Flow + +```mermaid +sequenceDiagram + participant GG as Greengrass Nucleus + participant WC as Workflow_Component + participant WD as WorkflowWatcher (LocalServer) + participant EX as WorkflowExecutor (LocalServer) + participant GM as GstPipelineManager (existing) + participant TR as Triton (existing) + + GG->>WC: deploy artifacts to /aws_dda/workflows/{workflowId}/{version}/ + WD->>WD: detect manifest.json (filesystem watch + startup scan) + WD->>EX: register workflow as runnable + Note over EX: trigger fires (interval / folder / MQTT / API) + EX->>EX: render pipeline string from compiled document
prepend component plugin dir to GST_PLUGIN_PATH + EX->>GM: run_pipeline(pipeline_string) + GM->>TR: emltriton inference call + GM-->>EX: tags / error + EX->>EX: post-pipeline outputs (MQTT, OPC UA)
status via existing reporting path +``` + +LocalServer never restarts when a Workflow_Component is deployed or removed: the component's lifecycle is "install artifacts only" (no long-running process), and LocalServer's watcher picks up changes at runtime. Existing `Pipeline_Configuration` execution is a separate code path that is not touched (Requirement 13). + +### Research Summary + +- **React Flow** ([reactflow.dev](https://reactflow.dev)) supports custom node types with multiple typed handles, `isValidConnection` callbacks for port-type checks at drag time, and JSON-serializable state (`nodes`, `edges`, `viewport`) — a direct fit for Requirements 1 and 3. It is the de facto standard for React node editors (used by Langflow, n8n-style tools). +- **Lumeo's node reference** ([docs.lumeo.com/docs/node-reference](https://docs.lumeo.com/docs/node-reference)) informed the node taxonomy: sources, transforms (dewarp/rotate/crop), model inference, logic/filter, and outputs (MQTT, digital I/O, OPC UA, capture). The catalog below mirrors this shape while mapping onto DDA's existing GStreamer elements (`emltriton`, `emoutputevent`, `emlcapture`, `videocrop`, `videoflip`, `videoconvert`). +- **Existing executor contract** (`src/backend/gstreamer/gst_pipeline.py`): pipelines are executed via `Gst.parse_launch(pipeline_str)` with a GLib main loop, a 120 s watchdog, and tag parsing for `is_anomalous`/`confidence`. The compiler therefore targets a launch-string-renderable document. Branching uses named `tee` elements (`tee name=t ... t. ! queue ! ...`), which `parse_launch` supports and the existing builder already uses for outputs. +- **Bedrock Converse API** supports tool-use (function calling) for structured JSON output across Anthropic, Amazon Nova, and Meta models, making the model identifier configurable without per-model prompt surgery (Requirement 10.6). +- **Greengrass component model**: components with only `Artifacts` and no `Lifecycle/Run` step install files and stay `FINISHED`; deploying/removing them does not affect other components — the mechanism relied on for Requirement 13.3. + +## Components and Interfaces + +### 1. Workflow_Builder (Frontend) + +Location: `edge-cv-portal/frontend/src/pages/workflows/`, `edge-cv-portal/frontend/src/components/workflow-builder/` + +- **Canvas**: React Flow instance with custom node component rendering category color, title, ports (handles) labeled with port types, and inline validation badges. +- **Node_Palette**: left sidebar listing node types grouped by category (input, preprocessing, model inference, post-processing, output), sourced from the node catalog endpoint (`GET /api/v1/workflows/node-catalog`), draggable via HTML5 drag-and-drop onto the canvas (Requirement 1.1, 1.2). +- **Connection rules**: React Flow `isValidConnection` calls `arePortsCompatible(sourcePort, targetPort)`; incompatible attempts are rejected and a toast/tooltip shows the reason, e.g. "Cannot connect VideoFrames output to EventSignal input" (Requirement 1.4). +- **Config panel**: right sidebar showing the selected node's parameter schema rendered as CloudScape form controls; edits validated against parameter type/constraints with inline errors (Requirement 1.7, 1.8). Model inference node's `modelName` parameter is a select populated from the existing model registry API filtered by Use_Case (Requirement 2.6). Custom_Python_Node exposes a code editor field plus declared input/output port type pickers (Requirement 2.7). +- **Inline validation**: a lightweight TypeScript mirror of validator checks V4 (missing required parameters) and V5 (unreachable nodes) runs on every graph mutation; affected nodes get warning badges that clear when resolved (Requirement 1.9, 1.10). Full validation (all checks) is invoked explicitly via a Validate button calling the backend (Requirement 4.8, 4.9). +- **Actions toolbar**: Save (versioned), Validate, Test, Generate (chat panel), Package, Deploy — each gated by the user's role from the existing auth context (Requirement 11). +- **Chat panel**: collapsible panel for prompt-based generation; maintains a session id, renders generated workflows onto the canvas only after client-side parse + backend validation, and never clobbers the canvas on failure (Requirement 10.3, 10.4). +- **Test panel**: dataset picker/uploader and per-node test result display, marking stubbed nodes with a "simulated" badge and limitation text (Requirement 12.2, 12.8). + +### 2. Node Catalog (`workflow_core.catalog`) + +The catalog is data, not code: a list of `NodeTypeDescriptor` records (see Data Models) declaring ports, parameters, per-architecture GStreamer mappings, and hardware-dependence flags (Requirement 2.8). Initial catalog: + +| Category | Node type | Ports (in → out) | GStreamer mapping (sketch) | Hardware-dependent | +|---|---|---|---|---| +| input | `camera_source` | → VideoFrames | `appsrc`/`v4l2src`/CSI file source chains reused from existing builder logic | yes | +| input | `folder_source` | → VideoFrames | `filesrc ! (jpegparse ! jpegdec | pngdec) ! videoconvert` per-arch variants (JP6 PNG staging path) | no | +| input | `digital_input` | → EventSignal | GPIO poll adapter (executor-level, not a GStreamer element) | yes | +| preprocessing | `dewarp` | VideoFrames → VideoFrames | `opencv`-based `dewarp` plugin (packaged dependency) | no | +| preprocessing | `rotate` | VideoFrames → VideoFrames | `videoflip method=
+ 
| Home |
| Features |
Modules
+| module | description | stable version | +devel version | status |
|---|---|---|---|---|
| gstreamer | +core library and elements | +1.28.5 | +1.29.2 | +active |
| gst-plugins-good | +a set of good-quality plug-ins under our preferred license, LGPL | +1.28.5 | 1.29.2 | active |
| gst-plugins-ugly | +a set of good-quality plug-ins that might pose distribution problems | +1.28.5 | 1.29.2 | active |
| gst-plugins-bad | +a set of plug-ins that need more quality, testing or documentation | +1.28.5 | 1.29.2 | active |
| qt-gstreamer | +QtGStreamer | ++ N/A + | ++ N/A + | +abandoned |
maintenance
", + "| foo |
|---|
| bar |
| Module | Description |
|---|