chore: bump version to 0.9.0-rc2

Rc/sientiapde 2173 - Mover repeat_last_prediction do sientia-dataops-library para o laborious, com suporte a metadata_id

.claude/commands/opsx/apply.md

@@ -0,0 +1,152 @@
+---
+name: "OPSX: Apply"
+description: Implement tasks from an OpenSpec change (Experimental)
+category: Workflow
+tags: [workflow, artifacts, experimental]
+---
+
+Implement tasks from an OpenSpec change.
+
+**Input**: Optionally specify a change name (e.g., `/opsx:apply add-auth`). If omitted, check if it can be inferred from conversation context. If vague or ambiguous you MUST prompt for available changes.
+
+**Steps**
+
+1. **Select the change**
+
+   If a name is provided, use it. Otherwise:
+   - Infer from conversation context if the user mentioned a change
+   - Auto-select if only one active change exists
+   - If ambiguous, run `openspec list --json` to get available changes and use the **AskUserQuestion tool** to let the user select
+
+   Always announce: "Using change: <name>" and how to override (e.g., `/opsx:apply <other>`).
+
+2. **Check status to understand the schema**
+   ```bash
+   openspec status --change "<name>" --json
+   ```
+   Parse the JSON to understand:
+   - `schemaName`: The workflow being used (e.g., "spec-driven")
+   - Which artifact contains the tasks (typically "tasks" for spec-driven, check status for others)
+
+3. **Get apply instructions**
+
+   ```bash
+   openspec instructions apply --change "<name>" --json
+   ```
+
+   This returns:
+   - `contextFiles`: artifact ID -> array of concrete file paths (varies by schema)
+   - Progress (total, complete, remaining)
+   - Task list with status
+   - Dynamic instruction based on current state
+
+   **Handle states:**
+   - If `state: "blocked"` (missing artifacts): show message, suggest using `/opsx:continue`
+   - If `state: "all_done"`: congratulate, suggest archive
+   - Otherwise: proceed to implementation
+
+4. **Read context files**
+
+   Read every file path listed under `contextFiles` from the apply instructions output.
+   The files depend on the schema being used:
+   - **spec-driven**: proposal, specs, design, tasks
+   - Other schemas: follow the contextFiles from CLI output
+
+5. **Show current progress**
+
+   Display:
+   - Schema being used
+   - Progress: "N/M tasks complete"
+   - Remaining tasks overview
+   - Dynamic instruction from CLI
+
+6. **Implement tasks (loop until done or blocked)**
+
+   For each pending task:
+   - Show which task is being worked on
+   - Make the code changes required
+   - Keep changes minimal and focused
+   - Mark task complete in the tasks file: `- [ ]` → `- [x]`
+   - Continue to next task
+
+   **Pause if:**
+   - Task is unclear → ask for clarification
+   - Implementation reveals a design issue → suggest updating artifacts
+   - Error or blocker encountered → report and wait for guidance
+   - User interrupts
+
+7. **On completion or pause, show status**
+
+   Display:
+   - Tasks completed this session
+   - Overall progress: "N/M tasks complete"
+   - If all done: suggest archive
+   - If paused: explain why and wait for guidance
+
+**Output During Implementation**
+
+```
+## Implementing: <change-name> (schema: <schema-name>)
+
+Working on task 3/7: <task description>
+[...implementation happening...]
+✓ Task complete
+
+Working on task 4/7: <task description>
+[...implementation happening...]
+✓ Task complete
+```
+
+**Output On Completion**
+
+```
+## Implementation Complete
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Progress:** 7/7 tasks complete ✓
+
+### Completed This Session
+- [x] Task 1
+- [x] Task 2
+...
+
+All tasks complete! You can archive this change with `/opsx:archive`.
+```
+
+**Output On Pause (Issue Encountered)**
+
+```
+## Implementation Paused
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Progress:** 4/7 tasks complete
+
+### Issue Encountered
+<description of the issue>
+
+**Options:**
+1. <option 1>
+2. <option 2>
+3. Other approach
+
+What would you like to do?
+```
+
+**Guardrails**
+- Keep going through tasks until done or blocked
+- Always read context files before starting (from the apply instructions output)
+- If task is ambiguous, pause and ask before implementing
+- If implementation reveals issues, pause and suggest artifact updates
+- Keep code changes minimal and scoped to each task
+- Update task checkbox immediately after completing each task
+- Pause on errors, blockers, or unclear requirements - don't guess
+- Use contextFiles from CLI output, don't assume specific file names
+
+**Fluid Workflow Integration**
+
+This skill supports the "actions on a change" model:
+
+- **Can be invoked anytime**: Before all artifacts are done (if tasks exist), after partial implementation, interleaved with other actions
+- **Allows artifact updates**: If implementation reveals design issues, suggest updating artifacts - not phase-locked, work fluidly

.claude/commands/opsx/archive.md

@@ -0,0 +1,157 @@
+---
+name: "OPSX: Archive"
+description: Archive a completed change in the experimental workflow
+category: Workflow
+tags: [workflow, archive, experimental]
+---
+
+Archive a completed change in the experimental workflow.
+
+**Input**: Optionally specify a change name after `/opsx:archive` (e.g., `/opsx:archive add-auth`). If omitted, check if it can be inferred from conversation context. If vague or ambiguous you MUST prompt for available changes.
+
+**Steps**
+
+1. **If no change name provided, prompt for selection**
+
+   Run `openspec list --json` to get available changes. Use the **AskUserQuestion tool** to let the user select.
+
+   Show only active changes (not already archived).
+   Include the schema used for each change if available.
+
+   **IMPORTANT**: Do NOT guess or auto-select a change. Always let the user choose.
+
+2. **Check artifact completion status**
+
+   Run `openspec status --change "<name>" --json` to check artifact completion.
+
+   Parse the JSON to understand:
+   - `schemaName`: The workflow being used
+   - `artifacts`: List of artifacts with their status (`done` or other)
+
+   **If any artifacts are not `done`:**
+   - Display warning listing incomplete artifacts
+   - Prompt user for confirmation to continue
+   - Proceed if user confirms
+
+3. **Check task completion status**
+
+   Read the tasks file (typically `tasks.md`) to check for incomplete tasks.
+
+   Count tasks marked with `- [ ]` (incomplete) vs `- [x]` (complete).
+
+   **If incomplete tasks found:**
+   - Display warning showing count of incomplete tasks
+   - Prompt user for confirmation to continue
+   - Proceed if user confirms
+
+   **If no tasks file exists:** Proceed without task-related warning.
+
+4. **Assess delta spec sync state**
+
+   Check for delta specs at `openspec/changes/<name>/specs/`. If none exist, proceed without sync prompt.
+
+   **If delta specs exist:**
+   - Compare each delta spec with its corresponding main spec at `openspec/specs/<capability>/spec.md`
+   - Determine what changes would be applied (adds, modifications, removals, renames)
+   - Show a combined summary before prompting
+
+   **Prompt options:**
+   - If changes needed: "Sync now (recommended)", "Archive without syncing"
+   - If already synced: "Archive now", "Sync anyway", "Cancel"
+
+   If user chooses sync, use Task tool (subagent_type: "general-purpose", prompt: "Use Skill tool to invoke openspec-sync-specs for change '<name>'. Delta spec analysis: <include the analyzed delta spec summary>"). Proceed to archive regardless of choice.
+
+5. **Perform the archive**
+
+   Create the archive directory if it doesn't exist:
+   ```bash
+   mkdir -p openspec/changes/archive
+   ```
+
+   Generate target name using current date: `YYYY-MM-DD-<change-name>`
+
+   **Check if target already exists:**
+   - If yes: Fail with error, suggest renaming existing archive or using different date
+   - If no: Move the change directory to archive
+
+   ```bash
+   mv openspec/changes/<name> openspec/changes/archive/YYYY-MM-DD-<name>
+   ```
+
+6. **Display summary**
+
+   Show archive completion summary including:
+   - Change name
+   - Schema that was used
+   - Archive location
+   - Spec sync status (synced / sync skipped / no delta specs)
+   - Note about any warnings (incomplete artifacts/tasks)
+
+**Output On Success**
+
+```
+## Archive Complete
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Archived to:** openspec/changes/archive/YYYY-MM-DD-<name>/
+**Specs:** ✓ Synced to main specs
+
+All artifacts complete. All tasks complete.
+```
+
+**Output On Success (No Delta Specs)**
+
+```
+## Archive Complete
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Archived to:** openspec/changes/archive/YYYY-MM-DD-<name>/
+**Specs:** No delta specs
+
+All artifacts complete. All tasks complete.
+```
+
+**Output On Success With Warnings**
+
+```
+## Archive Complete (with warnings)
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Archived to:** openspec/changes/archive/YYYY-MM-DD-<name>/
+**Specs:** Sync skipped (user chose to skip)
+
+**Warnings:**
+- Archived with 2 incomplete artifacts
+- Archived with 3 incomplete tasks
+- Delta spec sync was skipped (user chose to skip)
+
+Review the archive if this was not intentional.
+```
+
+**Output On Error (Archive Exists)**
+
+```
+## Archive Failed
+
+**Change:** <change-name>
+**Target:** openspec/changes/archive/YYYY-MM-DD-<name>/
+
+Target archive directory already exists.
+
+**Options:**
+1. Rename the existing archive
+2. Delete the existing archive if it's a duplicate
+3. Wait until a different date to archive
+```
+
+**Guardrails**
+- Always prompt for change selection if not provided
+- Use artifact graph (openspec status --json) for completion checking
+- Don't block archive on warnings - just inform and confirm
+- Preserve .openspec.yaml when moving to archive (it moves with the directory)
+- Show clear summary of what happened
+- If sync is requested, use the Skill tool to invoke `openspec-sync-specs` (agent-driven)
+- If delta specs exist, always run the sync assessment and show the combined summary before prompting

.claude/commands/opsx/explore.md

@@ -0,0 +1,173 @@
+---
+name: "OPSX: Explore"
+description: "Enter explore mode - think through ideas, investigate problems, clarify requirements"
+category: Workflow
+tags: [workflow, explore, experimental, thinking]
+---
+
+Enter explore mode. Think deeply. Visualize freely. Follow the conversation wherever it goes.
+
+**IMPORTANT: Explore mode is for thinking, not implementing.** You may read files, search code, and investigate the codebase, but you must NEVER write code or implement features. If the user asks you to implement something, remind them to exit explore mode first and create a change proposal. You MAY create OpenSpec artifacts (proposals, designs, specs) if the user asks—that's capturing thinking, not implementing.
+
+**This is a stance, not a workflow.** There are no fixed steps, no required sequence, no mandatory outputs. You're a thinking partner helping the user explore.
+
+**Input**: The argument after `/opsx:explore` is whatever the user wants to think about. Could be:
+- A vague idea: "real-time collaboration"
+- A specific problem: "the auth system is getting unwieldy"
+- A change name: "add-dark-mode" (to explore in context of that change)
+- A comparison: "postgres vs sqlite for this"
+- Nothing (just enter explore mode)
+
+---
+
+## The Stance
+
+- **Curious, not prescriptive** - Ask questions that emerge naturally, don't follow a script
+- **Open threads, not interrogations** - Surface multiple interesting directions and let the user follow what resonates. Don't funnel them through a single path of questions.
+- **Visual** - Use ASCII diagrams liberally when they'd help clarify thinking
+- **Adaptive** - Follow interesting threads, pivot when new information emerges
+- **Patient** - Don't rush to conclusions, let the shape of the problem emerge
+- **Grounded** - Explore the actual codebase when relevant, don't just theorize
+
+---
+
+## What You Might Do
+
+Depending on what the user brings, you might:
+
+**Explore the problem space**
+- Ask clarifying questions that emerge from what they said
+- Challenge assumptions
+- Reframe the problem
+- Find analogies
+
+**Investigate the codebase**
+- Map existing architecture relevant to the discussion
+- Find integration points
+- Identify patterns already in use
+- Surface hidden complexity
+
+**Compare options**
+- Brainstorm multiple approaches
+- Build comparison tables
+- Sketch tradeoffs
+- Recommend a path (if asked)
+
+**Visualize**
+```
+┌─────────────────────────────────────────┐
+│     Use ASCII diagrams liberally        │
+├─────────────────────────────────────────┤
+│                                         │
+│      ┌────────┐         ┌────────┐      │
+│      │ State  │────────▶│ State  │      │
+│      │   A    │         │   B    │      │
+│      └────────┘         └────────┘      │
+│                                         │
+│   System diagrams, state machines,      │
+│   data flows, architecture sketches,    │
+│   dependency graphs, comparison tables  │
+│                                         │
+└─────────────────────────────────────────┘
+```
+
+**Surface risks and unknowns**
+- Identify what could go wrong
+- Find gaps in understanding
+- Suggest spikes or investigations
+
+---
+
+## OpenSpec Awareness
+
+You have full context of the OpenSpec system. Use it naturally, don't force it.
+
+### Check for context
+
+At the start, quickly check what exists:
+```bash
+openspec list --json
+```
+
+This tells you:
+- If there are active changes
+- Their names, schemas, and status
+- What the user might be working on
+
+If the user mentioned a specific change name, read its artifacts for context.
+
+### When no change exists
+
+Think freely. When insights crystallize, you might offer:
+
+- "This feels solid enough to start a change. Want me to create a proposal?"
+- Or keep exploring - no pressure to formalize
+
+### When a change exists
+
+If the user mentions a change or you detect one is relevant:
+
+1. **Read existing artifacts for context**
+   - `openspec/changes/<name>/proposal.md`
+   - `openspec/changes/<name>/design.md`
+   - `openspec/changes/<name>/tasks.md`
+   - etc.
+
+2. **Reference them naturally in conversation**
+   - "Your design mentions using Redis, but we just realized SQLite fits better..."
+   - "The proposal scopes this to premium users, but we're now thinking everyone..."
+
+3. **Offer to capture when decisions are made**
+
+    | Insight Type               | Where to Capture               |
+    |----------------------------|--------------------------------|
+    | New requirement discovered | `specs/<capability>/spec.md` |
+    | Requirement changed        | `specs/<capability>/spec.md` |
+    | Design decision made       | `design.md`                  |
+    | Scope changed              | `proposal.md`                |
+    | New work identified        | `tasks.md`                   |
+    | Assumption invalidated     | Relevant artifact              |
+
+   Example offers:
+   - "That's a design decision. Capture it in design.md?"
+   - "This is a new requirement. Add it to specs?"
+   - "This changes scope. Update the proposal?"
+
+4. **The user decides** - Offer and move on. Don't pressure. Don't auto-capture.
+
+---
+
+## What You Don't Have To Do
+
+- Follow a script
+- Ask the same questions every time
+- Produce a specific artifact
+- Reach a conclusion
+- Stay on topic if a tangent is valuable
+- Be brief (this is thinking time)
+
+---
+
+## Ending Discovery
+
+There's no required ending. Discovery might:
+
+- **Flow into a proposal**: "Ready to start? I can create a change proposal."
+- **Result in artifact updates**: "Updated design.md with these decisions"
+- **Just provide clarity**: User has what they need, moves on
+- **Continue later**: "We can pick this up anytime"
+
+When things crystallize, you might offer a summary - but it's optional. Sometimes the thinking IS the value.
+
+---
+
+## Guardrails
+
+- **Don't implement** - Never write code or implement features. Creating OpenSpec artifacts is fine, writing application code is not.
+- **Don't fake understanding** - If something is unclear, dig deeper
+- **Don't rush** - Discovery is thinking time, not task time
+- **Don't force structure** - Let patterns emerge naturally
+- **Don't auto-capture** - Offer to save insights, don't just do it
+- **Do visualize** - A good diagram is worth many paragraphs
+- **Do explore the codebase** - Ground discussions in reality
+- **Do question assumptions** - Including the user's and your own

.claude/commands/opsx/propose.md

@@ -0,0 +1,106 @@
+---
+name: "OPSX: Propose"
+description: Propose a new change - create it and generate all artifacts in one step
+category: Workflow
+tags: [workflow, artifacts, experimental]
+---
+
+Propose a new change - create the change and generate all artifacts in one step.
+
+I'll create a change with artifacts:
+- proposal.md (what & why)
+- design.md (how)
+- tasks.md (implementation steps)
+
+When ready to implement, run /opsx:apply
+
+---
+
+**Input**: The argument after `/opsx:propose` is the change name (kebab-case), OR a description of what the user wants to build.
+
+**Steps**
+
+1. **If no input provided, ask what they want to build**
+
+   Use the **AskUserQuestion tool** (open-ended, no preset options) to ask:
+   > "What change do you want to work on? Describe what you want to build or fix."
+
+   From their description, derive a kebab-case name (e.g., "add user authentication" → `add-user-auth`).
+
+   **IMPORTANT**: Do NOT proceed without understanding what the user wants to build.
+
+2. **Create the change directory**
+   ```bash
+   openspec new change "<name>"
+   ```
+   This creates a scaffolded change at `openspec/changes/<name>/` with `.openspec.yaml`.
+
+3. **Get the artifact build order**
+   ```bash
+   openspec status --change "<name>" --json
+   ```
+   Parse the JSON to get:
+   - `applyRequires`: array of artifact IDs needed before implementation (e.g., `["tasks"]`)
+   - `artifacts`: list of all artifacts with their status and dependencies
+
+4. **Create artifacts in sequence until apply-ready**
+
+   Use the **TodoWrite tool** to track progress through the artifacts.
+
+   Loop through artifacts in dependency order (artifacts with no pending dependencies first):
+
+   a. **For each artifact that is `ready` (dependencies satisfied)**:
+      - Get instructions:
+        ```bash
+        openspec instructions <artifact-id> --change "<name>" --json
+        ```
+      - The instructions JSON includes:
+        - `context`: Project background (constraints for you - do NOT include in output)
+        - `rules`: Artifact-specific rules (constraints for you - do NOT include in output)
+        - `template`: The structure to use for your output file
+        - `instruction`: Schema-specific guidance for this artifact type
+        - `outputPath`: Where to write the artifact
+        - `dependencies`: Completed artifacts to read for context
+      - Read any completed dependency files for context
+      - Create the artifact file using `template` as the structure
+      - Apply `context` and `rules` as constraints - but do NOT copy them into the file
+      - Show brief progress: "Created <artifact-id>"
+
+   b. **Continue until all `applyRequires` artifacts are complete**
+      - After creating each artifact, re-run `openspec status --change "<name>" --json`
+      - Check if every artifact ID in `applyRequires` has `status: "done"` in the artifacts array
+      - Stop when all `applyRequires` artifacts are done
+
+   c. **If an artifact requires user input** (unclear context):
+      - Use **AskUserQuestion tool** to clarify
+      - Then continue with creation
+
+5. **Show final status**
+   ```bash
+   openspec status --change "<name>"
+   ```
+
+**Output**
+
+After completing all artifacts, summarize:
+- Change name and location
+- List of artifacts created with brief descriptions
+- What's ready: "All artifacts created! Ready for implementation."
+- Prompt: "Run `/opsx:apply` to start implementing."
+
+**Artifact Creation Guidelines**
+
+- Follow the `instruction` field from `openspec instructions` for each artifact type
+- The schema defines what each artifact should contain - follow it
+- Read dependency artifacts for context before creating new ones
+- Use `template` as the structure for your output file - fill in its sections
+- **IMPORTANT**: `context` and `rules` are constraints for YOU, not content for the file
+  - Do NOT copy `<context>`, `<rules>`, `<project_context>` blocks into the artifact
+  - These guide what you write, but should never appear in the output
+
+**Guardrails**
+- Create ALL artifacts needed for implementation (as defined by schema's `apply.requires`)
+- Always read dependency artifacts before creating a new one
+- If context is critically unclear, ask the user - but prefer making reasonable decisions to keep momentum
+- If a change with that name already exists, ask if user wants to continue it or create a new one
+- Verify each artifact file exists after writing before proceeding to next

.claude/skills/openspec-apply-change/SKILL.md

@@ -0,0 +1,156 @@
+---
+name: openspec-apply-change
+description: Implement tasks from an OpenSpec change. Use when the user wants to start implementing, continue implementation, or work through tasks.
+license: MIT
+compatibility: Requires openspec CLI.
+metadata:
+  author: openspec
+  version: "1.0"
+  generatedBy: "1.3.1"
+---
+
+Implement tasks from an OpenSpec change.
+
+**Input**: Optionally specify a change name. If omitted, check if it can be inferred from conversation context. If vague or ambiguous you MUST prompt for available changes.
+
+**Steps**
+
+1. **Select the change**
+
+   If a name is provided, use it. Otherwise:
+   - Infer from conversation context if the user mentioned a change
+   - Auto-select if only one active change exists
+   - If ambiguous, run `openspec list --json` to get available changes and use the **AskUserQuestion tool** to let the user select
+
+   Always announce: "Using change: <name>" and how to override (e.g., `/opsx:apply <other>`).
+
+2. **Check status to understand the schema**
+   ```bash
+   openspec status --change "<name>" --json
+   ```
+   Parse the JSON to understand:
+   - `schemaName`: The workflow being used (e.g., "spec-driven")
+   - Which artifact contains the tasks (typically "tasks" for spec-driven, check status for others)
+
+3. **Get apply instructions**
+
+   ```bash
+   openspec instructions apply --change "<name>" --json
+   ```
+
+   This returns:
+   - `contextFiles`: artifact ID -> array of concrete file paths (varies by schema - could be proposal/specs/design/tasks or spec/tests/implementation/docs)
+   - Progress (total, complete, remaining)
+   - Task list with status
+   - Dynamic instruction based on current state
+
+   **Handle states:**
+   - If `state: "blocked"` (missing artifacts): show message, suggest using openspec-continue-change
+   - If `state: "all_done"`: congratulate, suggest archive
+   - Otherwise: proceed to implementation
+
+4. **Read context files**
+
+   Read every file path listed under `contextFiles` from the apply instructions output.
+   The files depend on the schema being used:
+   - **spec-driven**: proposal, specs, design, tasks
+   - Other schemas: follow the contextFiles from CLI output
+
+5. **Show current progress**
+
+   Display:
+   - Schema being used
+   - Progress: "N/M tasks complete"
+   - Remaining tasks overview
+   - Dynamic instruction from CLI
+
+6. **Implement tasks (loop until done or blocked)**
+
+   For each pending task:
+   - Show which task is being worked on
+   - Make the code changes required
+   - Keep changes minimal and focused
+   - Mark task complete in the tasks file: `- [ ]` → `- [x]`
+   - Continue to next task
+
+   **Pause if:**
+   - Task is unclear → ask for clarification
+   - Implementation reveals a design issue → suggest updating artifacts
+   - Error or blocker encountered → report and wait for guidance
+   - User interrupts
+
+7. **On completion or pause, show status**
+
+   Display:
+   - Tasks completed this session
+   - Overall progress: "N/M tasks complete"
+   - If all done: suggest archive
+   - If paused: explain why and wait for guidance
+
+**Output During Implementation**
+
+```
+## Implementing: <change-name> (schema: <schema-name>)
+
+Working on task 3/7: <task description>
+[...implementation happening...]
+✓ Task complete
+
+Working on task 4/7: <task description>
+[...implementation happening...]
+✓ Task complete
+```
+
+**Output On Completion**
+
+```
+## Implementation Complete
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Progress:** 7/7 tasks complete ✓
+
+### Completed This Session
+- [x] Task 1
+- [x] Task 2
+...
+
+All tasks complete! Ready to archive this change.
+```
+
+**Output On Pause (Issue Encountered)**
+
+```
+## Implementation Paused
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Progress:** 4/7 tasks complete
+
+### Issue Encountered
+<description of the issue>
+
+**Options:**
+1. <option 1>
+2. <option 2>
+3. Other approach
+
+What would you like to do?
+```
+
+**Guardrails**
+- Keep going through tasks until done or blocked
+- Always read context files before starting (from the apply instructions output)
+- If task is ambiguous, pause and ask before implementing
+- If implementation reveals issues, pause and suggest artifact updates
+- Keep code changes minimal and scoped to each task
+- Update task checkbox immediately after completing each task
+- Pause on errors, blockers, or unclear requirements - don't guess
+- Use contextFiles from CLI output, don't assume specific file names
+
+**Fluid Workflow Integration**
+
+This skill supports the "actions on a change" model:
+
+- **Can be invoked anytime**: Before all artifacts are done (if tasks exist), after partial implementation, interleaved with other actions
+- **Allows artifact updates**: If implementation reveals design issues, suggest updating artifacts - not phase-locked, work fluidly

.claude/skills/openspec-archive-change/SKILL.md

@@ -0,0 +1,114 @@
+---
+name: openspec-archive-change
+description: Archive a completed change in the experimental workflow. Use when the user wants to finalize and archive a change after implementation is complete.
+license: MIT
+compatibility: Requires openspec CLI.
+metadata:
+  author: openspec
+  version: "1.0"
+  generatedBy: "1.3.1"
+---
+
+Archive a completed change in the experimental workflow.
+
+**Input**: Optionally specify a change name. If omitted, check if it can be inferred from conversation context. If vague or ambiguous you MUST prompt for available changes.
+
+**Steps**
+
+1. **If no change name provided, prompt for selection**
+
+   Run `openspec list --json` to get available changes. Use the **AskUserQuestion tool** to let the user select.
+
+   Show only active changes (not already archived).
+   Include the schema used for each change if available.
+
+   **IMPORTANT**: Do NOT guess or auto-select a change. Always let the user choose.
+
+2. **Check artifact completion status**
+
+   Run `openspec status --change "<name>" --json` to check artifact completion.
+
+   Parse the JSON to understand:
+   - `schemaName`: The workflow being used
+   - `artifacts`: List of artifacts with their status (`done` or other)
+
+   **If any artifacts are not `done`:**
+   - Display warning listing incomplete artifacts
+   - Use **AskUserQuestion tool** to confirm user wants to proceed
+   - Proceed if user confirms
+
+3. **Check task completion status**
+
+   Read the tasks file (typically `tasks.md`) to check for incomplete tasks.
+
+   Count tasks marked with `- [ ]` (incomplete) vs `- [x]` (complete).
+
+   **If incomplete tasks found:**
+   - Display warning showing count of incomplete tasks
+   - Use **AskUserQuestion tool** to confirm user wants to proceed
+   - Proceed if user confirms
+
+   **If no tasks file exists:** Proceed without task-related warning.
+
+4. **Assess delta spec sync state**
+
+   Check for delta specs at `openspec/changes/<name>/specs/`. If none exist, proceed without sync prompt.
+
+   **If delta specs exist:**
+   - Compare each delta spec with its corresponding main spec at `openspec/specs/<capability>/spec.md`
+   - Determine what changes would be applied (adds, modifications, removals, renames)
+   - Show a combined summary before prompting
+
+   **Prompt options:**
+   - If changes needed: "Sync now (recommended)", "Archive without syncing"
+   - If already synced: "Archive now", "Sync anyway", "Cancel"
+
+   If user chooses sync, use Task tool (subagent_type: "general-purpose", prompt: "Use Skill tool to invoke openspec-sync-specs for change '<name>'. Delta spec analysis: <include the analyzed delta spec summary>"). Proceed to archive regardless of choice.
+
+5. **Perform the archive**
+
+   Create the archive directory if it doesn't exist:
+   ```bash
+   mkdir -p openspec/changes/archive
+   ```
+
+   Generate target name using current date: `YYYY-MM-DD-<change-name>`
+
+   **Check if target already exists:**
+   - If yes: Fail with error, suggest renaming existing archive or using different date
+   - If no: Move the change directory to archive
+
+   ```bash
+   mv openspec/changes/<name> openspec/changes/archive/YYYY-MM-DD-<name>
+   ```
+
+6. **Display summary**
+
+   Show archive completion summary including:
+   - Change name
+   - Schema that was used
+   - Archive location
+   - Whether specs were synced (if applicable)
+   - Note about any warnings (incomplete artifacts/tasks)
+
+**Output On Success**
+
+```
+## Archive Complete
+
+**Change:** <change-name>
+**Schema:** <schema-name>
+**Archived to:** openspec/changes/archive/YYYY-MM-DD-<name>/
+**Specs:** ✓ Synced to main specs (or "No delta specs" or "Sync skipped")
+
+All artifacts complete. All tasks complete.
+```
+
+**Guardrails**
+- Always prompt for change selection if not provided
+- Use artifact graph (openspec status --json) for completion checking
+- Don't block archive on warnings - just inform and confirm
+- Preserve .openspec.yaml when moving to archive (it moves with the directory)
+- Show clear summary of what happened
+- If sync is requested, use openspec-sync-specs approach (agent-driven)
+- If delta specs exist, always run the sync assessment and show the combined summary before prompting

.claude/skills/openspec-explore/SKILL.md

@@ -0,0 +1,288 @@
+---
+name: openspec-explore
+description: Enter explore mode - a thinking partner for exploring ideas, investigating problems, and clarifying requirements. Use when the user wants to think through something before or during a change.
+license: MIT
+compatibility: Requires openspec CLI.
+metadata:
+  author: openspec
+  version: "1.0"
+  generatedBy: "1.3.1"
+---
+
+Enter explore mode. Think deeply. Visualize freely. Follow the conversation wherever it goes.
+
+**IMPORTANT: Explore mode is for thinking, not implementing.** You may read files, search code, and investigate the codebase, but you must NEVER write code or implement features. If the user asks you to implement something, remind them to exit explore mode first and create a change proposal. You MAY create OpenSpec artifacts (proposals, designs, specs) if the user asks—that's capturing thinking, not implementing.
+
+**This is a stance, not a workflow.** There are no fixed steps, no required sequence, no mandatory outputs. You're a thinking partner helping the user explore.
+
+---
+
+## The Stance
+
+- **Curious, not prescriptive** - Ask questions that emerge naturally, don't follow a script
+- **Open threads, not interrogations** - Surface multiple interesting directions and let the user follow what resonates. Don't funnel them through a single path of questions.
+- **Visual** - Use ASCII diagrams liberally when they'd help clarify thinking
+- **Adaptive** - Follow interesting threads, pivot when new information emerges
+- **Patient** - Don't rush to conclusions, let the shape of the problem emerge
+- **Grounded** - Explore the actual codebase when relevant, don't just theorize
+
+---
+
+## What You Might Do
+
+Depending on what the user brings, you might:
+
+**Explore the problem space**
+- Ask clarifying questions that emerge from what they said
+- Challenge assumptions
+- Reframe the problem
+- Find analogies
+
+**Investigate the codebase**
+- Map existing architecture relevant to the discussion
+- Find integration points
+- Identify patterns already in use
+- Surface hidden complexity
+
+**Compare options**
+- Brainstorm multiple approaches
+- Build comparison tables
+- Sketch tradeoffs
+- Recommend a path (if asked)
+
+**Visualize**
+```
+┌─────────────────────────────────────────┐
+│     Use ASCII diagrams liberally        │
+├─────────────────────────────────────────┤
+│                                         │
+│      ┌────────┐         ┌────────┐      │
+│      │ State  │────────▶│ State  │      │
+│      │   A    │         │   B    │      │
+│      └────────┘         └────────┘      │
+│                                         │
+│   System diagrams, state machines,      │
+│   data flows, architecture sketches,    │
+│   dependency graphs, comparison tables  │
+│                                         │
+└─────────────────────────────────────────┘
+```
+
+**Surface risks and unknowns**
+- Identify what could go wrong
+- Find gaps in understanding
+- Suggest spikes or investigations
+
+---
+
+## OpenSpec Awareness
+
+You have full context of the OpenSpec system. Use it naturally, don't force it.
+
+### Check for context
+
+At the start, quickly check what exists:
+```bash
+openspec list --json
+```
+
+This tells you:
+- If there are active changes
+- Their names, schemas, and status
+- What the user might be working on
+
+### When no change exists
+
+Think freely. When insights crystallize, you might offer:
+
+- "This feels solid enough to start a change. Want me to create a proposal?"
+- Or keep exploring - no pressure to formalize
+
+### When a change exists
+
+If the user mentions a change or you detect one is relevant:
+
+1. **Read existing artifacts for context**
+   - `openspec/changes/<name>/proposal.md`
+   - `openspec/changes/<name>/design.md`
+   - `openspec/changes/<name>/tasks.md`
+   - etc.
+
+2. **Reference them naturally in conversation**
+   - "Your design mentions using Redis, but we just realized SQLite fits better..."
+   - "The proposal scopes this to premium users, but we're now thinking everyone..."
+
+3. **Offer to capture when decisions are made**
+
+    | Insight Type               | Where to Capture               |
+    |----------------------------|--------------------------------|
+    | New requirement discovered | `specs/<capability>/spec.md` |
+    | Requirement changed        | `specs/<capability>/spec.md` |
+    | Design decision made       | `design.md`                  |
+    | Scope changed              | `proposal.md`                |
+    | New work identified        | `tasks.md`                   |
+    | Assumption invalidated     | Relevant artifact              |
+
+   Example offers:
+   - "That's a design decision. Capture it in design.md?"
+   - "This is a new requirement. Add it to specs?"
+   - "This changes scope. Update the proposal?"
+
+4. **The user decides** - Offer and move on. Don't pressure. Don't auto-capture.
+
+---
+
+## What You Don't Have To Do
+
+- Follow a script
+- Ask the same questions every time
+- Produce a specific artifact
+- Reach a conclusion
+- Stay on topic if a tangent is valuable
+- Be brief (this is thinking time)
+
+---
+
+## Handling Different Entry Points
+
+**User brings a vague idea:**
+```
+User: I'm thinking about adding real-time collaboration
+
+You: Real-time collab is a big space. Let me think about this...
+
+      COLLABORATION SPECTRUM
+      ════════════════════════════════════════════
+
+      Awareness          Coordination         Sync
+          │                   │                 │
+          ▼                   ▼                 ▼
+      ┌────────┐         ┌────────┐        ┌────────┐
+      │Presence│         │Cursors │        │  CRDT  │
+      │  "3    │         │ Multi  │        │Conflict│
+      │online" │         │ select │        │  free  │
+      └────────┘         └────────┘        └────────┘
+          │                   │                 │
+       trivial            moderate           complex
+
+      Where's your head at?
+```
+
+**User brings a specific problem:**
+```
+User: The auth system is a mess
+
+You: [reads codebase]
+
+     ┌─────────────────────────────────────────────┐
+     │              CURRENT AUTH FLOW              │
+     └─────────────────────────────────────────────┘
+                        │
+          ┌─────────────┼─────────────┐
+          ▼             ▼             ▼
+     ┌─────────┐  ┌─────────┐  ┌─────────┐
+     │ Google  │  │  GitHub │  │  Email  │
+     │  OAuth  │  │  OAuth  │  │  Magic  │
+     └────┬────┘  └────┬────┘  └────┬────┘
+          │             │             │
+          └─────────────┼─────────────┘
+                        ▼
+                  ┌───────────┐
+                  │  Session  │
+                  └─────┬─────┘
+                        │
+                        ▼
+                  ┌───────────┐
+                  │   Perms   │
+                  └───────────┘
+
+     I see three tangles. Which one's burning?
+```
+
+**User is stuck mid-implementation:**
+```
+User: /opsx:explore add-auth-system
+      The OAuth integration is more complex than expected
+
+You: [reads change artifacts]
+
+     You're on task 4: "Implement OAuth flow"
+
+     Let me trace what's involved...
+
+     [draws diagram, explores options, suggests paths]
+
+     Want to update the design to reflect this?
+     Or add a spike task to investigate?
+```
+
+**User wants to compare options:**
+```
+User: Should we use Postgres or SQLite?
+
+You: Generic answer is boring. What's the context?
+
+User: A CLI tool that tracks local dev environments
+
+You: That changes everything.
+
+     ┌─────────────────────────────────────────────────┐
+     │          CLI TOOL DATA STORAGE                  │
+     └─────────────────────────────────────────────────┘
+
+     Key constraints:
+     • No daemon running
+     • Must work offline
+     • Single user
+
+                  SQLite          Postgres
+     Deployment   embedded ✓      needs server ✗
+     Offline      yes ✓           no ✗
+     Single file  yes ✓           no ✗
+
+     SQLite. Not even close.
+
+     Unless... is there a sync component?
+```
+
+---
+
+## Ending Discovery
+
+There's no required ending. Discovery might:
+
+- **Flow into a proposal**: "Ready to start? I can create a change proposal."
+- **Result in artifact updates**: "Updated design.md with these decisions"
+- **Just provide clarity**: User has what they need, moves on
+- **Continue later**: "We can pick this up anytime"
+
+When it feels like things are crystallizing, you might summarize:
+
+```
+## What We Figured Out
+
+**The problem**: [crystallized understanding]
+
+**The approach**: [if one emerged]
+
+**Open questions**: [if any remain]
+
+**Next steps** (if ready):
+- Create a change proposal
+- Keep exploring: just keep talking
+```
+
+But this summary is optional. Sometimes the thinking IS the value.
+
+---
+
+## Guardrails
+
+- **Don't implement** - Never write code or implement features. Creating OpenSpec artifacts is fine, writing application code is not.
+- **Don't fake understanding** - If something is unclear, dig deeper
+- **Don't rush** - Discovery is thinking time, not task time
+- **Don't force structure** - Let patterns emerge naturally
+- **Don't auto-capture** - Offer to save insights, don't just do it
+- **Do visualize** - A good diagram is worth many paragraphs
+- **Do explore the codebase** - Ground discussions in reality
+- **Do question assumptions** - Including the user's and your own

.claude/skills/openspec-propose/SKILL.md

@@ -0,0 +1,110 @@
+---
+name: openspec-propose
+description: Propose a new change with all artifacts generated in one step. Use when the user wants to quickly describe what they want to build and get a complete proposal with design, specs, and tasks ready for implementation.
+license: MIT
+compatibility: Requires openspec CLI.
+metadata:
+  author: openspec
+  version: "1.0"
+  generatedBy: "1.3.1"
+---
+
+Propose a new change - create the change and generate all artifacts in one step.
+
+I'll create a change with artifacts:
+- proposal.md (what & why)
+- design.md (how)
+- tasks.md (implementation steps)
+
+When ready to implement, run /opsx:apply
+
+---
+
+**Input**: The user's request should include a change name (kebab-case) OR a description of what they want to build.
+
+**Steps**
+
+1. **If no clear input provided, ask what they want to build**
+
+   Use the **AskUserQuestion tool** (open-ended, no preset options) to ask:
+   > "What change do you want to work on? Describe what you want to build or fix."
+
+   From their description, derive a kebab-case name (e.g., "add user authentication" → `add-user-auth`).
+
+   **IMPORTANT**: Do NOT proceed without understanding what the user wants to build.
+
+2. **Create the change directory**
+   ```bash
+   openspec new change "<name>"
+   ```
+   This creates a scaffolded change at `openspec/changes/<name>/` with `.openspec.yaml`.
+
+3. **Get the artifact build order**
+   ```bash
+   openspec status --change "<name>" --json
+   ```
+   Parse the JSON to get:
+   - `applyRequires`: array of artifact IDs needed before implementation (e.g., `["tasks"]`)
+   - `artifacts`: list of all artifacts with their status and dependencies
+
+4. **Create artifacts in sequence until apply-ready**
+
+   Use the **TodoWrite tool** to track progress through the artifacts.
+
+   Loop through artifacts in dependency order (artifacts with no pending dependencies first):
+
+   a. **For each artifact that is `ready` (dependencies satisfied)**:
+      - Get instructions:
+        ```bash
+        openspec instructions <artifact-id> --change "<name>" --json
+        ```
+      - The instructions JSON includes:
+        - `context`: Project background (constraints for you - do NOT include in output)
+        - `rules`: Artifact-specific rules (constraints for you - do NOT include in output)
+        - `template`: The structure to use for your output file
+        - `instruction`: Schema-specific guidance for this artifact type
+        - `outputPath`: Where to write the artifact
+        - `dependencies`: Completed artifacts to read for context
+      - Read any completed dependency files for context
+      - Create the artifact file using `template` as the structure
+      - Apply `context` and `rules` as constraints - but do NOT copy them into the file
+      - Show brief progress: "Created <artifact-id>"
+
+   b. **Continue until all `applyRequires` artifacts are complete**
+      - After creating each artifact, re-run `openspec status --change "<name>" --json`
+      - Check if every artifact ID in `applyRequires` has `status: "done"` in the artifacts array
+      - Stop when all `applyRequires` artifacts are done
+
+   c. **If an artifact requires user input** (unclear context):
+      - Use **AskUserQuestion tool** to clarify
+      - Then continue with creation
+
+5. **Show final status**
+   ```bash
+   openspec status --change "<name>"
+   ```
+
+**Output**
+
+After completing all artifacts, summarize:
+- Change name and location
+- List of artifacts created with brief descriptions
+- What's ready: "All artifacts created! Ready for implementation."
+- Prompt: "Run `/opsx:apply` or ask me to implement to start working on the tasks."
+
+**Artifact Creation Guidelines**
+
+- Follow the `instruction` field from `openspec instructions` for each artifact type
+- The schema defines what each artifact should contain - follow it
+- Read dependency artifacts for context before creating new ones
+- Use `template` as the structure for your output file - fill in its sections
+- **IMPORTANT**: `context` and `rules` are constraints for YOU, not content for the file
+  - Do NOT copy `<context>`, `<rules>`, `<project_context>` blocks into the artifact
+  - These guide what you write, but should never appear in the output
+
+**Guardrails**
+- Create ALL artifacts needed for implementation (as defined by schema's `apply.requires`)
+- Always read dependency artifacts before creating a new one
+- If context is critically unclear, ask the user - but prefer making reasonable decisions to keep momentum
+- If a change with that name already exists, ask if user wants to continue it or create a new one
+- Verify each artifact file exists after writing before proceeding to next

.env.example

@@ -48,4 +48,4 @@ MINIO_ENDPOINT_URL="http://localhost:9000"
 MINIO_ACCESS_KEY="sientia"
 MINIO_SECRET_KEY="sientia"
 MINIO_REGION_NAME="sa-east-1"
-MINIO_DEFAULT_BUCKET="sientia"
+MINIO_DEFAULT_BUCKET="sientia"

.github/workflows/deploy.yml

@@ -0,0 +1,35 @@
+name: Deploy DataOps Worker
+
+on:
+  pull_request:
+    types: [closed]
+    branches: [main, 'release/**', 'feature/**']
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  id-token: write
+  pull-requests: read
+
+jobs:
+  deploy:
+    name: Deploy
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
+    uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
+    with:
+      branch_name: ${{ github.event.pull_request.head.ref }}
+      project_name: 'laborious'
+      chart_name: 'laborious-worker'
+      # Only the "legacy" runtime, for now (card 13 correction) — laborious-worker has two
+      # Deployments (basic, legacy). basic's GITHUB_BRANCH lives in the chart base (.global.env)
+      # and is deliberately left untouched here. legacy's own override exists ONLY at the profile
+      # level — profiles/gcp.yaml's laborious-worker.runtimes.legacy.env, not the chart base at
+      # all — hence profile/profile_env_list_path below instead of env_list_path.
+      profile: 'gcp'
+      profile_env_list_path: '[".runtimes.legacy.env"]'
+      update_version_in: '["pyproject"]'
+      create_tarball: ${{ github.event_name == 'pull_request' }}
+      mirror: true
+      bump_chart: true
+      deploy: true
+    secrets: inherit

.github/workflows/quality-gate.yml

@@ -1,17 +1,31 @@
 name: Quality gate
 
 on:
+  push:
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
   pull_request:
     branches:
       - main
+      - 'release/**'
+      - 'feature/**'
     types: [ opened, synchronize, reopened ]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   quality-gate:
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main
-    permissions: write-all
+    uses: Aignosi/github_workflow_templates/.github/workflows/python-quality-gate.yml@main
+    permissions:
+      contents: read
+      pull-requests: write
+      issues: write
     with:
       project_name: 'laborious'
-      repositories: 'sientia-dataops-library, sientia-mlops-library'
-      requirements_file: 'requirements-light.txt'
+      repositories: 'sientia-dataops-library,sientia-model-library'
+      requirements_file: 'requirements.txt'
     secrets: inherit

.github/workflows/release.yml

@@ -1,25 +0,0 @@
-name: Create Release on Merge to Main
-
-on:
-  pull_request:
-    types: [closed]
-    branches:
-      - main
-  workflow_dispatch:
-    inputs:
-      version:
-        description: 'Version to release'
-        required: false
-        type: string
-
-jobs:
-  release:
-    if: |
-      (github.event_name == 'pull_request' && github.event.pull_request.merged == true) ||
-      github.event_name == 'workflow_dispatch'
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main
-    permissions: write-all
-    with:
-      project_name: 'laborious'
-      release_version: ${{ github.event.inputs.version || '' }}
-    secrets: inherit

.github/workflows/require-chart-ref-label.yml

@@ -0,0 +1,16 @@
+name: Require chart-ref label
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
+
+permissions:
+  pull-requests: write
+
+jobs:
+  require-chart-ref-label:
+    uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

.gitignore

@@ -54,5 +54,6 @@ catboost_info/
 mlruns/
 
 relatorio*
-openspec/*
-.cursor/*
+openspec/
+.cursor/
+.claude/

Makefile

@@ -1,7 +0,0 @@
-VERSION = 1.0.8
-name = sientia-laborious
-# ENVIRONMENT = production
-
-docker-hub:
-	@docker build --no-cache -t aignosi.azurecr.io/$(name):$(VERSION) .
-	@docker push aignosi.azurecr.io/$(name):$(VERSION)

README.md

@@ -43,6 +43,7 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc
 - [Testing](#testing)
   - [Test Structure](#test-structure)
   - [Test Execution](#test-execution)
+  - [End-to-End Tests](#end-to-end-tests-e2e)
 - [Monitoring and Metrics](#monitoring-and-metrics)
   - [Application Health Metrics](#application-health-metrics)
   - [Prediction Operation Metrics](#prediction-operation-metrics)
@@ -126,10 +127,20 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
 ### Key Components
 
 #### **Worker (`laborious/worker/worker.py`)**
-- Temporal client setup, worker lifecycle, task queues
+- Temporal client setup, four workers via `sientia_do.temporal.worker.prepare_worker`
+- `RUNTIME` is required: installs the PluginStore runtime and suffixes the model-loading queues
 - Metrics server initialization, notification handler setup
 - Graceful shutdown and autoscaling-friendly behavior
 
+Task queues, one per worker:
+
+| Queue | Worker |
+|---|---|
+| `predictions_batch-{RUNTIME}-queue` | PredictionsBatch (+ its two sub-workflows) |
+| `minimal_retrain-{RUNTIME}-queue` | MinimalRetrain |
+| `drift-core-queue` | Drift (pinned to `core`: no model code loaded) |
+| `simple_metrics-core-queue` | SimpleMetrics (pinned to `core`: no model code loaded) |
+
 #### **Workflows (`laborious/workflows/`)**
 - `predictions_batch.py`: Batch prediction entry point
 - `sub_workflows/prediction_process.py`: Core prediction pipeline
@@ -584,10 +595,10 @@ Use the provided script to run the application locally:
 
 ```bash
 # Make script executable (first time only)
-chmod +x run_local.sh
+chmod +x scripts/run_local.sh
 
-# Run the application
-./run_local.sh
+# Run the application (from the repository root)
+./scripts/run_local.sh
 ```
 
 The script will:
@@ -601,10 +612,10 @@ Use the provided script to run tests with coverage:
 
 ```bash
 # Make script executable (first time only)
-chmod +x run_coverage.sh
+chmod +x scripts/run_coverage.sh
 
-# Run tests with coverage
-./run_coverage.sh
+# Run tests with coverage (from the repository root)
+./scripts/run_coverage.sh
 ```
 
 The script will:
@@ -755,6 +766,53 @@ pytest tests/laborious/activities/test_gates.py
 pytest tests/laborious/workflows/test_predictions_batch.py
 ```
 
+### End-to-End Tests (`e2e/`)
+
+The `e2e/` suite runs against real dependencies in Docker containers via
+[testcontainers](https://testcontainers.com/): PostgreSQL and MinIO. It lives outside `testpaths`,
+so it is never collected by the unit gate.
+
+**One-time setup, required before running anything under `e2e/`:**
+
+```bash
+echo "ryuk.disabled=true" >> ~/.testcontainers.properties
+```
+
+Without it, **every** e2e test fails at setup with a Docker mount error and never reaches an
+assertion:
+
+```
+docker.errors.APIError: 500 Server Error ... /containers/<id>/start:
+Mounts denied: The path /socket_mnt/home/<user>/.docker/desktop/docker.sock
+is not shared from the host and is not known to Docker.
+```
+
+The cause is not the tests. Before starting any test container, testcontainers starts a reaper
+container ("Ryuk") that bind-mounts the Docker socket so it can clean up if the test process dies
+without teardown. Under Docker Desktop on Linux that socket is `~/.docker/desktop/docker.sock`, a path
+Docker Desktop's File Sharing does not expose, so the reaper's own container cannot start. Note the
+symptom is `ERROR`, not `FAILED` — the failure is in fixture setup.
+
+Disabling it costs nothing in a normal run: every fixture stops its own containers in teardown. Ryuk
+only matters when the process is killed outright (`kill -9`, a crash, a hard stop from the IDE). After
+one of those, clean up with:
+
+```bash
+docker ps -a --filter "label=org.testcontainers=true" -q | xargs -r docker rm -f
+```
+
+**Running it:**
+
+```bash
+source ./venv/bin/activate
+
+# the predictions and OPC suites
+pytest e2e/ --override-ini testpaths=e2e -m integration
+```
+
+Full scenario catalogue:
+[`e2e/scenarios.md`](e2e/scenarios.md).
+
 ## 📊 Monitoring and Metrics
 
 The Laborious system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring:
@@ -801,6 +859,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
 |----------|-------------|---------|----------|
 | `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes |
 | `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No |
+| `RUNTIME` | Task queue suffix for all workflows (`{workflow}-{RUNTIME}-queue`) | _(none)_ | Yes |
 | `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
 | `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes |
 | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
@@ -845,7 +904,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
 | `PROJECT_NAME` | Project name for metrics | `laborious` | No |
 | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
 | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
-| `POD_ID` | Kubernetes pod identifier | `None` | No |
+| `HOSTNAME` | Pod identifier (set by Kubernetes) used in metrics labels | `None` | No |
 | `MINIO_ENDPOINT_URL` | MinIO endpoint URL | `http://localhost:9000` | Yes |
 | `MINIO_ACCESS_KEY` | MinIO access key | `minioadmin` | Yes |
 | `MINIO_SECRET_KEY` | MinIO secret key | `minioadmin` | Yes |
@@ -1233,7 +1292,9 @@ export LOG_LEVEL=DEBUG
 ### Scaling Considerations
 
 - **Horizontal Scaling**: Deploy multiple worker instances
-- **Task Queue Distribution**: Use multiple task queues for different workflow types
+- **Task Queue Distribution**: One worker pod per `RUNTIME`; queues are
+  `predictions_batch-{RUNTIME}-queue`, `minimal_retrain-{RUNTIME}-queue`,
+  `drift-core-queue`, `simple_metrics-core-queue`
 - **Database Performance**: Optimize indexes and connection pooling
 - **MLFlow Performance**: Configure appropriate model serving resources
 

e2e/conftest.py

@@ -29,6 +29,7 @@ from laborious.workflows.sub_workflows.prediction_process import PredictionProce
 from sientia_do.notifications.handlers import CoreNotificationHandler
 from sientia_do.observability.logger import Logger
 from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
 
 # Single source of truth for the test database schema. Mirrors the production
 # DDL for ``sientia_data`` so any production change can be pasted directly into
@@ -141,6 +142,24 @@ def notification_inserts(notification_handler):
         collection.insert_one = original_insert_one
 
 
+@pytest_asyncio.fixture
+def mongo_repository(mock_logger, notification_handler, metrics_controller, mongo_container):
+    """Real MongoDBRepository for gate metadata (prediction_metadata), same testcontainer as
+    ``notification_handler`` above."""
+    mongo_port = mongo_container.get_exposed_port(27017)
+    repository = MongoDBRepository(
+        connection_string=f'mongodb://localhost:{mongo_port}',
+        database_name='test_db',
+        logger=mock_logger,
+        notification_handler=notification_handler,
+        metrics_controller=metrics_controller,
+    )
+    try:
+        yield repository
+    finally:
+        repository.close()
+
+
 class _FakeModelWrapper:
     """External MLflow wrapper double used by repository stub."""
 
@@ -233,6 +252,7 @@ async def test_activities(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_repository_stub,
+    mongo_repository,
 ):
     """Activities with real infra and external-system stubs only."""
     minio_client = minio_container.get_client()
@@ -269,6 +289,7 @@ async def test_activities(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     activities.opc_repository = {'1': opc_repository_stub}
@@ -289,6 +310,7 @@ async def test_activities_real_minio(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_repository_stub,
+    mongo_repository,
 ):
     """Compatibility alias for offload tests."""
     minio_client = minio_container.get_client()
@@ -325,6 +347,7 @@ async def test_activities_real_minio(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     activities.opc_repository = {'1': opc_repository_stub}
@@ -535,6 +558,7 @@ async def test_activities_real_opc(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_e2e_server: OpcE2ETestServer,
+    mongo_repository,
 ):
     """Activities with real OpcRepository connected to the in-process OPC UA server."""
     minio_client = minio_container.get_client()
@@ -571,6 +595,7 @@ async def test_activities_real_opc(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     await asyncio.to_thread(_connect_activities_to_opc_server, activities, opc_e2e_server.url)

e2e/db_schema.sql

@@ -40,6 +40,7 @@ CREATE TABLE IF NOT EXISTS sientia_data.predictions (
     "timestamp"           timestamptz NOT NULL,
     created_at            timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
     "comments"            text NULL,
+    metadata_id           varchar(24) NULL,
     CONSTRAINT unique_model_id_timestamp
         UNIQUE (model_id, "timestamp")
 );

e2e/opc_test_server.py

@@ -147,8 +147,11 @@ class OpcE2ETestServer:
         value = await self._confidence_node.read_value()
         return float(value)
 
-    async def _pre_write_callback(self, _event, _service) -> None:
-        if self._session_bad_on_write:
+    async def _pre_write_callback(self, event, _service) -> None:
+        # Only client writes are faulted. asyncua's own clock task writes CurrentTime
+        # through this same PreWrite dispatch, and raising there kills its time_task;
+        # the stored exception then resurfaces as a teardown failure in server.stop().
+        if self._session_bad_on_write and getattr(event, 'is_external', False):
             raise ServiceError(ua.StatusCodes.BadSessionIdInvalid)
 
 

e2e/scenarios.md

@@ -7,10 +7,51 @@ It is a functional reference of scenario behavior, inputs, and expected outcomes
 
 ## Execution Context
 
+### One-time setup: disable the testcontainers reaper
+
+Do this once per machine, before running anything under `e2e/`:
+
+```bash
+echo "ryuk.disabled=true" >> ~/.testcontainers.properties
+```
+
+**Why.** Before starting any test container, testcontainers starts a reaper container ("Ryuk") whose
+job is to clean up containers if the test process dies without running its teardown. Ryuk bind-mounts
+the Docker socket. Under **Docker Desktop on Linux** the socket lives at
+`~/.docker/desktop/docker.sock`, which Docker Desktop's File Sharing does not expose, so the mount is
+denied and the reaper cannot start:
+
+```
+docker.errors.APIError: 500 Server Error ... /containers/<id>/start:
+Mounts denied: The path /socket_mnt/home/<user>/.docker/desktop/docker.sock
+is not shared from the host and is not known to Docker.
+```
+
+Every test then errors **at setup** — `ERROR`, not `FAILED`, and never inside an assertion. If you see
+that, this is why.
+
+**What you give up.** Nothing in a normal run: every fixture in `conftest.py` stops its own containers
+in teardown. Ryuk only covers the case where the process is killed without teardown (`kill -9`, a
+crash, a hard stop from the IDE). After one of those, clean up by hand:
+
+```bash
+docker ps -a --filter "label=org.testcontainers=true" -q | xargs -r docker rm -f
+```
+
+*Alternative, if you want to keep the reaper:* point Docker at the classic socket with
+`export DOCKER_HOST=unix:///var/run/docker.sock`. That only works if a native Docker Engine is
+running alongside Docker Desktop; if the Desktop daemon is the only one, disabling Ryuk is the way.
+
+
 - Tests run under `e2e/` and are marked with `@pytest.mark.integration`.
 - PostgreSQL and MinIO are provisioned with testcontainers.
 - `test_minio_offload.py` uses real MinIO I/O; other scenario suites may use stubs/mocks for optional outputs.
 - Real OPC UA scenarios use `@pytest.mark.opc` and an in-process asyncua server (`e2e/test_opc_real_server.py`).
+- **Nothing under `e2e/` is covered by `validate`**: that script lints `laborious/` and `tests/`
+  only, and its test step is `pytest tests/` (`testpaths = ["tests"]`). Lint these files by hand
+  (`ruff format --check e2e/ && ruff check e2e/`), and remember that `addopts` carries
+  `--strict-markers`: a test without `@pytest.mark.integration` is not collected, and a run that
+  collects nothing still exits 0. Check the collected count, not the exit code.
 
 ### Local validation
 

e2e/test_opc_real_server.py

@@ -7,13 +7,17 @@ Mock-based OPC tests remain in test_predictions_batch_format_export.py.
 
 import asyncio
 import threading
-import time
 
 import pytest
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
-from e2e.helpers import assert_prediction, insert_sample_data, make_workflow_id, start_and_await_workflow
+from e2e.helpers import (
+    assert_prediction,
+    insert_sample_data,
+    make_workflow_id,
+    start_and_await_workflow,
+)
 from e2e.opc_test_server import UNKNOWN_NODE_ID, OpcE2ETestServer, build_opc_output_config
 from e2e.test_predictions_batch_format_export import get_base_input_data
 from laborious.activities.activities import Activities
@@ -22,16 +26,24 @@ from laborious.utils.repository.opc_repository import OpcRepository
 from laborious.workflows.predictions_batch import PredictionsBatch
 
 
-def _slow_reconnect_under_lock(repo: OpcRepository, hold_seconds: float = 0.75) -> None:
+def _reconnect_under_held_lock(
+    repo: OpcRepository, acquired: threading.Event, release: threading.Event
+) -> None:
     """
-    Hold the connection lock briefly so concurrent writes see reconnect_in_progress.
+    Hold the connection lock until released, then reconnect.
+
+    The lock must stay held for the whole workflow, and how long the workflow takes to
+    reach the OPC write is not bounded — a fixed sleep here lets `_reconnect_locked`
+    finish first, which sets `_session_ready` again and the write then succeeds.
 
     Args:
         repo (OpcRepository): Connected repository.
-        hold_seconds (float): Time to keep the lock before reconnecting.
+        acquired (threading.Event): Set once the lock is held, before any write can race.
+        release (threading.Event): Waited on while holding the lock; set by the caller.
     """
     with repo._connection_lock:
-        time.sleep(hold_seconds)
+        acquired.set()
+        release.wait()
         repo._reconnect_locked()
 
 
@@ -172,12 +184,15 @@ async def test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server(
 
     repo = test_activities_real_opc.opc_repository['1']
     repo._session_ready.clear()
+    acquired = threading.Event()
+    release = threading.Event()
     reconnect_thread = threading.Thread(
-        target=_slow_reconnect_under_lock,
-        args=(repo,),
+        target=_reconnect_under_held_lock,
+        args=(repo, acquired, release),
         daemon=True,
     )
     reconnect_thread.start()
+    acquired.wait(timeout=5.0)
 
     input_data = get_base_input_data(model_id)
     input_data['opc_output_config'] = build_opc_output_config(opc_e2e_server.node_ids)
@@ -191,6 +206,7 @@ async def test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server(
             make_workflow_id('test-opc-real-reconnect-block'),
         )
     finally:
+        release.set()
         reconnect_thread.join(timeout=5.0)
 
     assert_prediction(

encode.sh

@@ -1,13 +0,0 @@
-source ./venv/bin/activate
-
-pip install pathspec
-pip install pyyaml
-
-echo "
-.git" >> .gitignore
-
-python encrypt.py ./ code --ignore .gitignore --chunk-size 100000
-
-sed -i '/.git/d' .gitignore
-
-xdg-open .

encrypt.py

@@ -1,113 +0,0 @@
-import os
-import argparse
-from pathspec import PathSpec
-import yaml # type: ignore
-from typing import Any
-
-'''
-Usage:
-    python .\encrypt.py path_to_dir output_file --ignore ignore_file --chunk-size 100000
-'''
-
-
-def load_ignore_patterns(ignore_file, include_library):
-    # Ensure the .gitignore file exists
-    if not os.path.exists(ignore_file):
-        raise FileNotFoundError(f"Ignore file not found at {ignore_file}")
-
-    # Load and parse the .gitignore patterns
-    with open(ignore_file, 'r') as file:
-        patterns = file.readlines()
-        if not include_library:
-            patterns.append('**/deploy/library/')
-
-        spec = PathSpec.from_lines('gitwildmatch', patterns)
-        return spec
-
-
-def is_ignored(file_path, spec):
-    """Check if a file should be ignored based on the ignore patterns."""
-    return spec.match_file(file_path) if spec else False
-
-
-def encode_file_tree_to_yaml(directory, ignore_file, include_library):
-    """Encode the file tree into a single YAML file."""
-    ignore_patterns = load_ignore_patterns(
-        ignore_file, include_library) if ignore_file else None
-    file_tree: dict[str, Any] = {}
-
-    for root, dirs, files in os.walk(directory):
-        # Skip ignored directories
-        dirs[:] = [d for d in dirs if not is_ignored(
-            os.path.join(root, d), ignore_patterns)]
-
-        for file in files:
-            file_path = os.path.join(root, file)
-
-            # Skip ignored files
-            if is_ignored(file_path, ignore_patterns):
-                continue
-
-            # Read file content
-            try:
-                with open(file_path, 'r', encoding='utf-8') as f:
-                    content = f.read()
-            except Exception as e:
-                print(f"Error reading file {file_path}: {e}")
-                raise
-
-            # Create nested dictionary structure
-            path_parts = os.path.relpath(file_path, directory).split(os.sep)
-            current_level = file_tree
-
-            # all except the last part (the file name)
-            for part in path_parts[:-1]:
-                current_level = current_level.setdefault(part, {})
-
-            # Add the file and its content
-            current_level[path_parts[-1]] = content
-    return yaml.dump(file_tree, default_flow_style=False)
-
-
-def chunk_and_write_file_tree_to_yaml(yaml_content, output_file, chunk_size=None):
-    """Chunk the YAML content and write it to the output file."""
-
-    chunks = [yaml_content] if chunk_size is None else [
-        yaml_content[i:i + chunk_size] for i in range(0, len(yaml_content), chunk_size)]
-
-    for i, chunk in enumerate(chunks):
-        chunk_file = f"{output_file}_{i}.yaml"
-        # Write the file tree to the output YAML file
-        with open(chunk_file, 'w', encoding='utf-8') as yaml_file:
-            yaml_file.write(chunk)
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Encrypts file tree to yaml file")
-    parser.add_argument("input_directory", help="Directory to encode")
-    parser.add_argument("output_yaml_file", help="Output YAML file")
-    parser.add_argument("--ignore", default=None,
-                        help="Path to the ignore file")
-    parser.add_argument("--chunk-size", type=int, default=None,
-                        help="Chunk size for the output YAML file")
-    parser.add_argument("--library", type=bool, default=False,
-                        help="Incude the library in the output YAML file")
-
-    # Parse arguments
-    args = parser.parse_args()
-
-    # Example usage
-    directory_to_encode = args.input_directory
-    ignore_file_path = args.ignore
-    output_yaml_file = args.output_yaml_file
-    include_library = args.library
-
-    content = encode_file_tree_to_yaml(
-        directory_to_encode, ignore_file_path, include_library)
-    chunk_and_write_file_tree_to_yaml(
-        content, output_yaml_file, args.chunk_size)
-
-
-if __name__ == "__main__":
-    main()

laborious/activities/activities.py

@@ -7,6 +7,8 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.observability.logger import Logger
     from sientia_do.repository.minio_repository_sync import MinioRepository
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
+    from sientia_do.utils.connectors_config import build_mongodb_config
     from sientia_model.model_repository.mlflow_repository import SientiaMLflowRepository
     from sientia_model.model_repository.plugin_store import PluginStore
 
@@ -15,18 +17,21 @@ with workflow.unsafe.imports_passed_through():
     from laborious.activities.mlflow import MLFlow
     from laborious.activities.model_metrics import ModelMetrics
     from laborious.activities.opc import OPC
+    from laborious.activities.prediction_metadata import PredictionMetadata
     from laborious.activities.storage import Storage
     from laborious.utils.connectors_config import build_mlflow_config
 
 
-class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
+class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetadata):
     """
     Central orchestrator for all Temporal activities used by Laborious workflows.
 
     Composes Storage (Postgres + MinIO offload), MLFlow (wrapper-based inference and retrain
     via ``SientiaMLflowRepository``), Gates (data quality and ML response filters), OPC exports,
-    drift/simple metrics, and PI Web API writes. The worker constructs one ``Activities`` instance
-    per process and registers its callables on multiple workers bound to different task queues.
+    drift/simple metrics, PI Web API writes, and PredictionMetadata (the single write point for
+    gate metadata — see `laborious.activities.prediction_metadata`). The worker constructs one
+    ``Activities`` instance per process and registers its callables on multiple workers bound to
+    different task queues.
 
     MLflow connectivity: unless ``mlflow_repository`` is injected (tests only), this class builds
     ``SientiaMLflowRepository`` from ``build_mlflow_config()`` so tracking credentials and URL
@@ -48,6 +53,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController | None = None,
         mlflow_repository: SientiaMLflowRepository | None = None,
+        mongo_repository: MongoDBRepository | None = None,
     ):
         """
         Wire Postgres, MinIO, MLflow, OPC, gates, metrics, and PI Web API into a single object.
@@ -67,6 +73,9 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
          - metrics_controller: Optional shared controller; if ``None``, a new one is created.
          - mlflow_repository: Optional ``SientiaMLflowRepository`` for unit/e2e tests; in production
            leave unset so the repository is built from environment via ``build_mlflow_config()``.
+         - mongo_repository: Optional ``MongoDBRepository`` (prediction metadata, best-effort) for
+           unit/e2e tests; in production leave unset so the repository is built from environment
+           via ``build_mongodb_config()``, same pattern as ``mlflow_repository`` above.
 
         Raises:
             Exception: If any parent ``__init__`` fails (e.g. invalid config keys).
@@ -89,6 +98,16 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
                 metrics_controller=mc,
             )
 
+        if mongo_repository is None:
+            mongo_cfg = build_mongodb_config()
+            mongo_repository = MongoDBRepository(
+                connection_string=mongo_cfg['connection_string'],
+                database_name=mongo_cfg['database_name'],
+                logger=logger,
+                notification_handler=notification_handler,
+                metrics_controller=mc,
+            )
+
         minio_repository = MinioRepository(
             endpoint=minio_config['endpoint_url'],
             access_key=minio_config['access_key'],
@@ -134,6 +153,14 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
             metrics_controller=mc,
         )
 
+        PredictionMetadata.__init__(
+            self,
+            mongo_repository=mongo_repository,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=mc,
+        )
+
         OPC.__init__(
             self,
             opc_servers=opc_config,
@@ -175,3 +202,4 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         OPC.close(self)
         ModelMetrics.close(self)
         API.close(self)
+        PredictionMetadata.close(self)

laborious/activities/gates.py

@@ -27,6 +27,7 @@ with workflow.unsafe.imports_passed_through():
 InputFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
 ResponseFilterFunc = Callable[[dict[str, Any], dict[str, Any]], bool]
 ContentFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
+InputCounterFunc = Callable[[DataFrame, dict[str, Any]], int]
 
 # Input filter function mappings
 input_filter_functions: dict[str, InputFilterFunc] = {
@@ -34,6 +35,42 @@ input_filter_functions: dict[str, InputFilterFunc] = {
     'EMPTY_DATA': filter_empty_data,
 }
 
+
+def _count_specific_variables_null_values(data: DataFrame, config: dict[str, Any]) -> int:
+    """
+    Count rows with a null value in one of the configured variables.
+
+    Mirrors `filter_specific_variables_null_values`'s own trigger condition, so the count
+    reflects exactly what made that filter fire.
+
+    Return:
+        int: Number of rows with a null value in `config['variables']`.
+    """
+    if data.empty:
+        return 0
+    return int((data['variable'].isin(config['variables']) & data['value'].isna()).sum())
+
+
+def _count_empty_data(_data: DataFrame, _config: dict[str, Any]) -> int:
+    """
+    Count rows for the EMPTY_DATA filter.
+
+    `EMPTY_DATA` only triggers when there are no rows at all, so the count is always 0.
+
+    Return:
+        int: Always 0.
+    """
+    return 0
+
+
+# Per-filter row counts for the `counter` field in prediction metadata (gate-local: mirrors each
+# filter's own trigger condition instead of changing `conditional_filters.py`'s bool contract,
+# which `mlflow_content_gate` also depends on for EMPTY_DATA).
+input_filter_counters: dict[str, InputCounterFunc] = {
+    'SPECIFIC_VARIABLES_NULL_VALUES': _count_specific_variables_null_values,
+    'EMPTY_DATA': _count_empty_data,
+}
+
 # Confidence mappings kept separate from function maps to avoid Union types
 input_path_confidence: Mapping[str, int] = {
     'STOP': -1,
@@ -163,7 +200,9 @@ class Gates(SientiaMonitoring):
         return policy, filter_config
 
     @activity.defn(name='input_gate')
-    def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+    def input_gate(
+        self, input_data: dict[str, Any]
+    ) -> tuple[str | None, int, str, dict[str, Any] | None]:
         """
         Apply input data quality filters and validation.
 
@@ -178,6 +217,9 @@ class Gates(SientiaMonitoring):
         3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
         4. Provides confidence scores and detailed comments
         5. Handles errors gracefully with notification integration
+        6. Builds a metadata document describing which filter(s) triggered, with their policy
+           and a counter, whenever one does — this activity only computes it, it never writes it
+           (see `PredictionMetadata.write_prediction_metadata` for the actual write)
 
         Args:
             input_data: Configuration and data for input validation
@@ -188,10 +230,14 @@ class Gates(SientiaMonitoring):
                     - path_priority (list[str]): Priority order for path decisions
 
         Returns:
-            tuple: (path_flag, confidence, comment)
+            tuple: (path_flag, confidence, comment, document)
                 - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
                 - confidence (int): Confidence score for the decision
                 - comment (str): Detailed explanation of the decision
+                - document (dict | None): Metadata describing the filter(s) that produced
+                    `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
+                    `None` when nothing was filtered. Not written anywhere by this activity —
+                    the caller decides whether and how to persist it.
 
         Raises:
             Exception: If filter execution fails or configuration is invalid
@@ -205,7 +251,8 @@ class Gates(SientiaMonitoring):
         data = payload.retrieve(self.minio_repository, metadata)
         path_priority = input_data['path_priority']
 
-        filter_output = []
+        # (filter_name, policy, filter_config) per filter that triggered, in configured order
+        triggered_filters: list[tuple[str, str, dict[str, Any]]] = []
 
         self._debug_dataframe('Input data:', data, metadata)
         self.debug(f'Filters: {filters}', metadata)
@@ -219,7 +266,7 @@ class Gates(SientiaMonitoring):
             try:
                 if input_filter_functions[fil](data, filter_config):
                     self.debug(f'Data not passed the input filter {fil}:{config}', metadata)
-                    filter_output.append(policy)
+                    triggered_filters.append((fil, policy, filter_config))
             except Exception as e:
                 trace = traceback.format_exc()
                 self.send_notification(
@@ -231,16 +278,46 @@ class Gates(SientiaMonitoring):
                     attachment_content=trace,
                 )
 
+        triggered_policies = [policy for _, policy, _ in triggered_filters]
+
         for path_flag in path_priority:
-            if path_flag in filter_output:
+            if path_flag in triggered_policies:
                 self.info(f'Input gate result: {path_flag}', metadata)
-                return path_flag, input_path_confidence[path_flag], 'Input data with bad quality'
+
+                matched = [
+                    (fil, filter_config)
+                    for fil, policy, filter_config in triggered_filters
+                    if policy == path_flag
+                ]
+                document = {
+                    'filters': [
+                        {
+                            'name': fil,
+                            'policy': path_flag,
+                            # `.get(..., lambda *_: 0)`, not `[fil]`: a filter added to
+                            # `input_filter_functions` without a matching counter here degrades to
+                            # counter=0 instead of a raw KeyError crashing this prediction (review
+                            # finding, 2026-09-30 — see test enforcing key parity between the two).
+                            'counter': input_filter_counters.get(fil, lambda *_: 0)(
+                                data, filter_config
+                            ),
+                        }
+                        for fil, filter_config in matched
+                    ],
+                }
+
+                return (
+                    path_flag,
+                    input_path_confidence[path_flag],
+                    'Input data with bad quality',
+                    document,
+                )
 
         self.info('Nothing was filtered by the input gate', metadata)
 
         del data
 
-        return None, 0, ''
+        return None, 0, '', None
 
     @activity.defn(name='mlflow_response_gate')
     def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
@@ -636,6 +713,9 @@ class Gates(SientiaMonitoring):
                 - model_id (str): Unique identifier for the ML model
                 - prediction_confidence (float): Confidence score (typically low for errors)
                 - comment (str): Error description or operational comment
+                - metadata_id (str | None, optional): Id of the `prediction_metadata` Mongo
+                    document to reference for this row; `None`/absent when there's nothing to
+                    reference
 
         Returns:
             dict: Formatted default prediction data with error indicators
@@ -653,6 +733,7 @@ class Gates(SientiaMonitoring):
                 'prediction_confidence': [input_data['prediction_confidence']],
                 'prediction_status': ['Bad'],
                 'comments': [input_data['comment']],
+                'metadata_id': [input_data.get('metadata_id')],
             }
         )
 

laborious/activities/mlflow.py

@@ -29,6 +29,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_model.model_repository.plugin_store import PluginStore
 
     from laborious.utils.dataframe_debug import build_dataframe_debug_message
+    from laborious.utils.messages import MISSING_MINIO_REPOSITORY
     from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
 
 
@@ -445,7 +446,7 @@ class MLFlow(SientiaMonitoring):
         """
 
         if self.minio_repository is None:
-            raise ValueError('Minio repository not initialized')
+            raise ValueError(MISSING_MINIO_REPOSITORY)
 
         metadata = input_data['metadata']
 
@@ -528,22 +529,23 @@ class MLFlow(SientiaMonitoring):
             )
 
             # Keep heavy model fitting outside MLflow run timing.
-            prediction_data = wrapper.retrain(data)
-            prediction_data.rename(columns={target: 'prediction'}, inplace=True)
+            prediction_data = wrapper.retrain(data).rename(columns={target: 'prediction'})
 
-            # Merge prediction data with retrain data
-            evaluation_data = pd.merge(
-                data, prediction_data, left_index=True, right_index=True, how='left'
+            # Both frames are indexed by the pivoted timestamp: expose it as an explicit column
+            # and join on it. `validate` fails loudly instead of silently multiplying rows if the
+            # wrapper ever returns duplicated timestamps.
+            evaluation_data = (
+                data.reset_index(names='timestamp')
+                .merge(
+                    prediction_data.reset_index(names='timestamp'),
+                    on='timestamp',
+                    how='left',
+                    validate='one_to_one',
+                )
+                .rename(columns={target: 'target'})
+                .sort_values(by='timestamp', ascending=True, ignore_index=True)
             )
 
-            # Rename target column to "target"
-            evaluation_data.rename(columns={target: 'target'}, inplace=True)
-
-            # Reset index and put as column "timestamp"
-            evaluation_data['timestamp'] = evaluation_data.index
-            evaluation_data.reset_index(drop=True, inplace=True)
-            evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True)
-
             run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}'
 
             with self.mlflow_repository.start_run(

laborious/activities/prediction_metadata.py

@@ -0,0 +1,132 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+
+    from bson import ObjectId
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.observability.logger import Logger
+    from sientia_do.observability.metrics_controller import MetricsController
+    from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
+
+
+class PredictionMetadata(SientiaMonitoring):
+    """
+    Centralized, best-effort writer for prediction gate metadata (triggered filter, policy,
+    counter) to MongoDB.
+
+    Gates only compute and return a metadata document; they never write it themselves. Keeping
+    the write here, instead of in each gate, puts the one real write point in the subworkflow
+    that's about to persist a `predictions` row — the only place that reliably knows whether one
+    is being written at all — and lets other metadata sources reuse the same activity instead of
+    duplicating write logic wherever they originate.
+    """
+
+    mongo_repository: MongoDBRepository | None = None
+    _PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
+
+    def __init__(
+        self,
+        mongo_repository: MongoDBRepository | None = None,
+        logger: Logger | None = None,
+        notification_handler: NotificationHandler | None = None,
+        metrics_controller: MetricsController | None = None,
+    ):
+        """
+        Initialize the prediction metadata writer.
+
+        Args:
+            mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
+                written best-effort. `None` is a valid state (see `write_prediction_metadata`).
+
+        Raises:
+            Exception: If BaseActivity initialization fails
+        """
+        self.mongo_repository = mongo_repository
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self) -> None:
+        """
+        Close the prediction metadata writer and clean up resources.
+        """
+        if self.mongo_repository is not None:
+            try:
+                self.mongo_repository.close()
+            finally:
+                self.mongo_repository = None
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        self.close()
+
+    @activity.defn(name='write_prediction_metadata')
+    def write_prediction_metadata(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Best-effort write of one or more gate metadata documents to Mongo, combined into a single
+        `prediction_metadata` document.
+
+        Implements both business rules for this feature in one place:
+        - The prediction write must never be blocked or failed by a metadata write. A failed
+          `insert(...)` is swallowed, never re-raised, logged here with call-site context (which
+          `mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s
+          `cleanup_minio_objects_expired` also logs again at the call site even though the
+          repository already logs internally and emits its own Prometheus metric
+          (`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this
+          one is for whoever is reading logs for this specific call.
+        - Nothing is written when there's nothing relevant to report. An empty/missing
+          `documents` list (or one full of falsy/empty entries) skips Mongo entirely, so callers
+          can call this unconditionally and get `None` back for "no metadata this time" instead
+          of deciding upfront whether there's something worth saving.
+
+        Takes a *list* of documents, not a single one: each caller contributes its own document
+        (shaped like `{'filters': [...]}`), and this flattens every `filters` entry across all of
+        them into one `prediction_metadata` document before the single insert. Today only one
+        gate ever contributes a document per prediction cycle (the gate chain short-circuits on
+        the first non-`None` `path_flag`), but the list shape means this activity won't need to
+        change again once more than one source contributes metadata for the same cycle.
+
+        `insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here,
+        client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this
+        codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char
+        size of the `predictions.metadata_id` Postgres column that will reference it.
+
+        Args:
+            input_data: Dict with:
+                - documents (list[dict[str, Any]]): Metadata payloads to combine, each shaped
+                  like `{'filters': [...]}`. Missing/empty, or entries with no `filters`, skip
+                  the write entirely.
+                - metadata (dict): Workflow metadata for contextual logging.
+
+        Return:
+            The generated id, or `None` when there's nothing to write, Mongo isn't configured, or
+            the insert failed.
+        """
+        metadata = input_data['metadata']
+        documents = input_data.get('documents') or []
+
+        filters = [
+            entry for document in documents if document for entry in document.get('filters', [])
+        ]
+
+        if not filters or self.mongo_repository is None:
+            return None
+
+        generated_id = str(ObjectId())
+
+        try:
+            self.mongo_repository.insert(
+                self._PREDICTION_METADATA_COLLECTION,
+                {'filters': filters, '_id': generated_id},
+                metadata,
+            )
+        except Exception as e:
+            self.error(f'Failed to write prediction metadata: {e}', metadata)
+            return None
+
+        return generated_id

laborious/activities/storage.py

@@ -2,6 +2,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     # Extend the Temporal Postgres activities for convenient query -> MinIO export
+    import time
     import traceback
     from datetime import timedelta
     from typing import Any
@@ -9,14 +10,18 @@ with workflow.unsafe.imports_passed_through():
     import pandas as pd
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability import metrics
     from sientia_do.observability.logger import Logger
     from sientia_do.observability.metrics_controller import MetricsController
     from sientia_do.observability.sientia_monitoring import SientiaMonitoring
     from sientia_do.repository.minio_repository_sync import MinioRepository
     from sientia_do.temporal.activities.postgres_sync import Postgres
     from sientia_do.temporal.constants import now
+    from sqlalchemy import text
 
+    from laborious.utils.messages import MISSING_MINIO_REPOSITORY
     from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
+    from laborious.utils.time import to_naive
 
 _LOAD_QUERY_OFFLOAD_SKIP_KEYS = frozenset({'model_name', 'key_prefix', 'size_threshold_bytes'})
 
@@ -24,7 +29,9 @@ _LOAD_QUERY_OFFLOAD_SKIP_KEYS = frozenset({'model_name', 'key_prefix', 'size_thr
 class Storage(Postgres, SientiaMonitoring):
     """
     Extensions for Postgres activities with a helper to export query results
-    directly to MinIO as Parquet and return the object name.
+    directly to MinIO as Parquet and return the object name, and with
+    ``repeat_last_prediction``, a query specific to this repo's REPEAT path
+    (moved here from the shared library — nothing else consumes it).
     """
 
     minio_repository: MinioRepository | None = None
@@ -84,7 +91,7 @@ class Storage(Postgres, SientiaMonitoring):
             dict[str, Any]: Flat ``MinioDataFramePayload`` dict or ``success: False`` on failure.
         """
         if self.minio_repository is None:
-            raise ValueError('Minio repository not initialized')
+            raise ValueError(MISSING_MINIO_REPOSITORY)
 
         metadata: dict = input_data.get('metadata', {})
         model_name = input_data['model_name']
@@ -139,6 +146,110 @@ class Storage(Postgres, SientiaMonitoring):
             }
         )
 
+    def _build_repeat_last_prediction_query(self, schema: str, table_name: str):
+        table_identifier = f'"{schema}".{table_name}'
+        return text(  # noqa: S608
+            f"""
+        INSERT INTO {table_identifier} (
+            model_id,
+            prediction,
+            timestamp,
+            response_time,
+            prediction_status,
+            prediction_confidence,
+            metadata_id,
+            created_at
+        )
+        SELECT
+            model_id,
+            prediction,
+            :last_timestamp,
+            response_time,
+            prediction_status,
+            prediction_confidence,
+            :metadata_id,
+            NOW()
+        FROM {table_identifier}
+        WHERE model_id = :model
+        ORDER BY timestamp DESC
+        LIMIT 1;
+        """  # nosec B608
+        )
+
+    @activity.defn(name='repeat_last_prediction')
+    def repeat_last_prediction(self, query_items: dict[str, Any]):
+        """
+        Insert a new `predictions` row that copies the last one for a model forward in time.
+
+        REPEAT persists a new row (unlike STOP, which persists nothing), reusing the previous
+        row's `prediction`/`response_time`/`prediction_status`/`prediction_confidence` values with
+        a fresh `timestamp`/`created_at`. `metadata_id` references the `prediction_metadata`
+        document for this cycle, if any (see `PredictionMetadata.write_prediction_metadata`);
+        `None` is a valid value — the column is nullable.
+
+        Args (query_items):
+            metadata (dict): Workflow metadata for logging and metrics.
+            schema (str): Postgres schema of the `predictions` table.
+            table_name (str): `predictions` table name.
+            model (str): `model_id` to repeat the last prediction for.
+            last_timestamp: Timestamp to stamp the new row with.
+            metadata_id (str | None): `prediction_metadata` document id, or `None`.
+
+        Raises:
+            Exception: Re-raised after logging/metrics if the insert fails.
+        """
+        metadata = query_items['metadata']
+        schema = query_items['schema']
+        table_name = query_items['table_name']
+        model = query_items['model']
+        last_timestamp = query_items['last_timestamp']
+        metadata_id = query_items.get('metadata_id')
+
+        repeat_query = self._build_repeat_last_prediction_query(schema, table_name)
+        self.info(f'Repeating last prediction for model {model}', metadata)
+        self.debug(
+            f'Query parameters: schema={schema}, model={model}, last_timestamp={last_timestamp}',
+            metadata,
+        )
+        core_labels = {
+            **self.get_core_labels(metadata, 'execute_sql'),
+            'conflict_mode': 'error',
+        }
+
+        with self.session_factory.begin() as session:
+            try:
+                start_time = time.time()
+                session.execute(
+                    repeat_query,
+                    {
+                        'last_timestamp': last_timestamp,
+                        'model': model,
+                        'metadata_id': metadata_id,
+                    },
+                )
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='ERROR_REPEATING_LAST_PREDICTION',
+                    message=f'Error repeating last prediction: {e}',
+                    block='repeat_last_prediction',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace,
+                )
+                self.error(trace, metadata)
+                self.emit_metric_sync(
+                    metric_object=metrics.POSTGRES_WRITE_ERROR_COUNT,
+                    tags=core_labels,
+                )
+                raise
+            else:
+                self.observe_lag_sync(start_time, metrics.POSTGRES_WRITE_LAG, core_labels)
+                self.emit_metric_sync(
+                    metric_object=metrics.POSTGRES_WRITE_COUNT,
+                    tags=core_labels,
+                )
+
     @activity.defn(name='cleanup_minio_objects_expired')
     def cleanup_minio_objects_expired(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
@@ -152,15 +263,13 @@ class Storage(Postgres, SientiaMonitoring):
             dict[str, Any]: ``success``, ``deleted_count``, and optional ``message``.
         """
         if self.minio_repository is None:
-            raise ValueError('Minio repository not initialized')
+            raise ValueError(MISSING_MINIO_REPOSITORY)
 
         metadata = input_data.get('metadata', {})
         payload = MinioDataFramePayload.from_dict(input_data['data'])
         prefix = payload.cleanup_prefix()
         base = now()
-        cutoff = (base.replace(tzinfo=None) if base.tzinfo else base) - timedelta(
-            hours=self.retention_hours
-        )
+        cutoff = to_naive(base) - timedelta(hours=self.retention_hours)
 
         report: dict[str, Any] = {
             'failed': {},
@@ -223,7 +332,8 @@ class Storage(Postgres, SientiaMonitoring):
         sessions, and monitoring hooks), then closes the optional MinIO repository
         and clears the local reference to avoid accidental reuse after shutdown.
         """
-        Postgres.close(self)
+        if hasattr(self, 'engine'):
+            Postgres.close(self)
         if self.minio_repository is not None:
             try:
                 self.minio_repository.close()

laborious/utils/messages.py

@@ -0,0 +1 @@
+MISSING_MINIO_REPOSITORY = 'Minio repository not initialized'

laborious/utils/repository/opc_repository.py

@@ -87,7 +87,7 @@ def _opc_authentication_token_str(client: Client | None) -> str:
         if tok is None:
             return 'unknown'
         return str(tok)
-    except Exception:
+    except Exception:  # best-effort label for logs/metrics, must not fail loud
         return 'unknown'
 
 
@@ -278,12 +278,14 @@ class OpcRepository(SientiaMonitoring):
         Configure certificates and timeouts on the sync asyncua client.
 
         Raises:
-            ValueError: If cert paths or client are missing.
+            ValueError: If cert paths, server URI or client are missing.
         """
         if self.cert_path is None or self.private_key_path is None:
             raise ValueError(
                 'Certificate and private key paths must be provided for secure connection.'
             )
+        if self.server_uri is None:
+            raise ValueError('server_uri must be provided for secure connection.')
 
         cert = Path(self.cert_path)
         private_key = Path(self.private_key_path)
@@ -841,7 +843,10 @@ class OpcRepository(SientiaMonitoring):
         session_id = _opc_authentication_token_str(self.client)
 
         try:
-            assert self.client is not None
+            if self.client is None:
+                raise OpcClientNotInitializedError(
+                    f'OPC client is not initialized for server {self.id}; cannot write {node}'
+                )
             node_obj = self.client.get_node(node)
         except Exception as e:
             if is_reconnectable_opcua_bad(e):

laborious/utils/time.py

@@ -0,0 +1,6 @@
+from datetime import datetime
+
+
+def to_naive(moment: datetime) -> datetime:
+    """Strip tzinfo, if any. Postgres `TIMESTAMP WITHOUT TIME ZONE` columns need naive values."""
+    return moment.replace(tzinfo=None) if moment.tzinfo else moment

laborious/worker/worker.py

@@ -8,8 +8,8 @@ repository, single PluginStore handle) but polls a different task queue.
 Task queues (see ``sientia_do.temporal.worker.prepare_worker``):
 - ``predictions_batch-{runtime}-queue`` + sub-workflows on the same queue (ML-heavy path).
 - ``minimal_retrain-{runtime}-queue`` (retrain + promote + export).
-- ``drift-queue`` and ``simple_metrics-queue`` without a runtime suffix so existing schedulers
-  keep stable queue names.
+- ``drift-core-queue`` and ``simple_metrics-core-queue``: pinned to the ``core`` runtime because
+  drift and metrics do not load model code, so every runtime shares the same queues.
 
 Bootstrap order:
 1. Prometheus app metrics and Mongo-backed notification handler.
@@ -24,7 +24,7 @@ Environment Variables:
 - RUNTIME: Required non-empty string passed to ``install_runtime``.
 - STORE_* / PYPI_*: Plugin store and private index (see ``build_plugin_store_config``).
 - TEMPORAL_HOST, TEMPORAL_NAMESPACE: Cluster connection.
-- POD_ID, HTTP_METRICS_PORT, HTTP_SDK_METRICS_PORT: Observability.
+- HOSTNAME (pod id), HTTP_METRICS_PORT, HTTP_SDK_METRICS_PORT: Observability.
 - PROJECT_NAME, MONGODB_*: Notifications (via ``build_mongodb_config`` in handler).
 - POSTGRES_*, MINIO_*, OPC_*, PI_WEB_API_*, MLFLOW_*: Passed through ``Activities`` helpers.
 """
@@ -75,14 +75,14 @@ async def main():
 
     Exits the process with code 0 on normal completion of all worker tasks, or 1 after logging
     if any worker raises. ``finally`` always shuts down notifications and activities and sets
-    ``app_up`` to 0 before ``sys.exit``.
+    ``app_up`` to 0; ``sys.exit`` runs after it.
 
     Raises:
         Exception: Propagated from ``asyncio.gather`` only before ``finally`` handling; typically
             workers run until cancelled.
 
     Return:
-        None (process terminates via ``sys.exit`` from the ``finally`` block).
+        None (process terminates via ``sys.exit`` after the ``finally`` block).
     """
     host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
     logger = get_logger(__name__)
@@ -98,7 +98,7 @@ async def main():
     logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata)
 
     logger.custom_info('Starting prometheus client...', metadata)
-    start_prometheus_server()
+    start_prometheus_server(logger, metadata)
 
     logger.custom_info('Starting Notification Handler...', metadata)
 
@@ -157,7 +157,7 @@ async def main():
         metrics.APP_UP.labels(pod_id=POD_ID).set(0)
         sys.exit(1)
 
-    logger.custom_info('Starting Activities...', metadata)
+    logger.custom_info('Starting Activities...', metadata_runtime)
 
     activities = Activities(
         postgres_config=build_postgres_config(),
@@ -170,10 +170,13 @@ async def main():
         metrics_controller=metrics_controller,
     )
 
-    logger.custom_info('Initializing OPC...', metadata)
+    logger.custom_info('Initializing OPC...', metadata_runtime)
     activities.init_opc()
 
-    logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
+    logger.custom_info(
+        f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...',
+        metadata_runtime,
+    )
 
     new_runtime = Runtime(
         telemetry=TelemetryConfig(
@@ -181,7 +184,7 @@ async def main():
         )
     )
 
-    logger.custom_info(f'Starting Temporal Client at {host}...', metadata)
+    logger.custom_info(f'Starting Temporal Client at {host}...', metadata_runtime)
 
     temporal_client = await client.Client.connect(
         target_host=host,
@@ -189,7 +192,7 @@ async def main():
         runtime=new_runtime,
     )
 
-    logger.custom_info('Starting Workers...', metadata)
+    logger.custom_info(f'Starting Workers (runtime={runtime})...', metadata_runtime)
 
     workers = [
         prepare_worker(
@@ -246,6 +249,8 @@ async def main():
                 activities.format_transformed_data,
                 activities.format_prediction,
                 activities.format_default_prediction,
+                # Prediction metadata
+                activities.write_prediction_metadata,
                 # OPC
                 activities.write_opc_data,
                 # Postgres / MinIO offload
@@ -267,22 +272,29 @@ async def main():
     for w in workers:
         handlers.append(w.run())
 
-    logger.custom_info('Workers started successfully', metadata)
+    logger.custom_info('Workers started successfully', metadata_runtime)
 
     exit_code = 0
     try:
         await asyncio.gather(*handlers)
-    except BaseException as e:  # NOSONAR
-        logger.custom_error(f'An unhandled exception occurred: {e}', metadata)
+    except Exception as e:
+        logger.custom_error(f'An unhandled exception occurred: {e}', metadata_runtime)
         exit_code = 1
     finally:
+        # No `except (SystemExit, KeyboardInterrupt)` on purpose: `finally` runs on every path,
+        # caught or not, so cleanup still happens for a deliberate shutdown (Ctrl+C, a handler's
+        # own sys.exit) without us having to catch and re-raise it ourselves — which would only
+        # let us relabel its exit code before `sys.exit(exit_code)` below discards it anyway.
+        # Letting it propagate on its own keeps the original code (e.g. a handler's sys.exit(N))
+        # intact instead of collapsing it to 0.
         notification_handler.shutdown()
         activities.shutdown()
         metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
-        sys.exit(exit_code)
+
+    sys.exit(exit_code)
 
 
-def start_prometheus_server():
+def start_prometheus_server(logger, metadata: dict):
     """
     Starts the Prometheus metrics server for monitoring and observability.
 
@@ -294,7 +306,7 @@ def start_prometheus_server():
 
     Environment Variables:
         HTTP_METRICS_PORT: Port for the metrics server (default: 9090)
-        POD_ID: Pod identifier for metrics labeling
+        HOSTNAME: Pod identifier for metrics labeling
 
     Raises:
         SystemExit: If the metrics server fails to start
@@ -302,10 +314,10 @@ def start_prometheus_server():
     try:
         port = int(os.getenv('HTTP_METRICS_PORT', 9090))
         start_http_server(port)
-        print(f'Prometheus server started on port {port}.')
+        logger.custom_info(f'Prometheus server started on port {port}.', metadata)
         metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
     except Exception as e:
-        print(f'Failed to start Prometheus server: {e}')
+        logger.custom_critical(f'Failed to start Prometheus server: {e}', metadata)
         os._exit(1)
 
 

laborious/workflows/drift.py

@@ -37,6 +37,7 @@ class Drift:
         model_config = input_data['model_config']
         target_name = model_config['target']
 
+        # Built from internal Temporal workflow config, not user input (bandit B608).
         gathering_query = f"""
             SELECT *
             FROM "{input_data['schema']}"."{input_data['source_table_name']}"
@@ -44,7 +45,7 @@ class Drift:
                 model_id = '{input_data['model_id']}' AND
                 timestamp > NOW() - INTERVAL '{input_data['interval']} minutes'
             ORDER BY timestamp ASC
-        """  # nosec B608 - values come from internal Temporal workflow config, not user input
+        """  # nosec
 
         target_data_handler = workflow.start_activity_method(
             Activities.load_custom_query,

laborious/workflows/minimal_retrain.py

@@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through():
     from typing import Any
 
     from sientia_do.temporal.policies import retry_policy
+    from temporalio.exceptions import ApplicationError
 
     from laborious.activities.activities import Activities
     from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
@@ -87,7 +88,11 @@ class MinimalRetrain:
 
         storage_payload = MinioDataFramePayload.from_dict(storage_result)
         if not storage_payload.has_data():
-            raise ValueError('No data returned from query')
+            raise ApplicationError(
+                'No data returned from query',
+                type='NoDataReturnedError',
+                non_retryable=True,
+            )
 
         experiment_response = await workflow.execute_activity_method(
             Activities.retrain_model,

laborious/workflows/simple_metrics.py

@@ -32,6 +32,7 @@ class SimpleMetrics:
         model_config = input_data['model_config']
         target_name = model_config['target']
 
+        # Built from internal Temporal workflow config, not user input (bandit B608).
         query = f"""
             select p."timestamp", p.prediction, ld.value as "target"
             from "{input_data['schema']}"."{input_data['predictions_table_name']}" p
@@ -45,7 +46,7 @@ class SimpleMetrics:
                 p."timestamp" >= NOW() - INTERVAL '{interval_minutes} minutes'
             order by
                 p."timestamp" desc;
-        """  # nosec B608 - values come from internal Temporal workflow config, not user input
+        """  # nosec
 
         target_data = await workflow.execute_activity_method(
             Activities.load_custom_query,

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -75,6 +75,11 @@ class FormatAndExportPrediction:
                         Required when path_flag is None
                     - comment (str): Operational comment or error description
                         Required when path_flag is not None
+                    - metadata_id (str | None): Id of the `prediction_metadata` Mongo document to
+                        reference for this row, already written by the caller; `None` when there's
+                        nothing to reference. Only meaningful when path_flag is not None — the
+                        normal (path_flag is None) path never has metadata to reference, so this
+                        key is irrelevant there.
 
         Returns:
             None: The workflow completes successfully when all export operations finish
@@ -154,6 +159,7 @@ class FormatAndExportPrediction:
                     'model_id': input_data['model_id'],
                     'prediction_confidence': prediction_confidence,
                     'comment': input_data['comment'],
+                    'metadata_id': input_data.get('metadata_id'),
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(seconds=60),

laborious/workflows/sub_workflows/prediction_process.py

@@ -132,7 +132,9 @@ class PredictionProcess:
             'path_priority': input_data['path_priority'],
         }
 
-        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+        # `document` describes which filter(s) produced `path_flag` (`None` when nothing was
+        # filtered); `path_flag_handler` is the one that decides whether and how to persist it.
+        path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
             Activities.input_gate,
             gate_input,
             retry_policy=retry_policy,
@@ -141,7 +143,7 @@ class PredictionProcess:
 
         # Handle path decision based on filter results
         if await self.path_flag_handler(
-            data, path_flag, input_data, confidence, last_timestamp, comment
+            data, path_flag, input_data, confidence, last_timestamp, comment, document
         ):
             return
 
@@ -247,6 +249,51 @@ class PredictionProcess:
             },
         )
 
+    async def _write_prediction_metadata(
+        self, metadata: dict[str, Any], document: dict[str, Any] | None
+    ) -> str | None:
+        """
+        Write a gate's metadata document via the centralized write activity.
+
+        Shared by the CONTINUE and REPEAT branches of `path_flag_handler` — both persist a new
+        `predictions` row (`REPEAT` reuses the previous row's values but still inserts a new
+        one), so both need a `metadata_id` to reference it.
+
+        Args:
+            metadata: Workflow metadata for contextual logging.
+            document: Metadata produced by the gate that resolved this `path_flag`; `None` when
+                nothing was filtered, or when the gate that produced this `path_flag` doesn't
+                build one yet.
+
+        Return:
+            The generated id, or `None` when there's nothing to write, including when the
+            activity call itself fails (see `PredictionMetadata.write_prediction_metadata`).
+        """
+        if not document:
+            # Skip the remote activity call entirely rather than dispatching it just to have
+            # `write_prediction_metadata` decide there's nothing to write — `mlflow_response_gate`
+            # and `mlflow_content_gate` don't build a `document` yet, so this is the common case
+            # today, not a rare edge case (review finding, 2026-10-01).
+            return None
+
+        try:
+            return await workflow.execute_activity_method(
+                Activities.write_prediction_metadata,
+                {**metadata, 'documents': [document]},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=1),
+            )
+        except Exception as e:
+            # Business rule 1: writing metadata must never block or fail the prediction write.
+            # `write_prediction_metadata` already swallows Mongo-specific failures internally and
+            # returns `None`, but that protection doesn't cover the activity call itself failing
+            # to run at all (its `retry_policy` has `maximum_attempts=1` — no retry — so a
+            # timeout, a momentarily unavailable worker, or any other dispatch failure raises
+            # straight into the workflow). Catching it here, instead of just at the Mongo client
+            # level, is what actually makes the rule hold (review finding, 2026-10-01).
+            workflow.logger.warning(f'Failed to write prediction metadata, continuing anyway: {e}')
+            return None
+
     async def path_flag_handler(
         self,
         data: dict[str, Any],
@@ -255,6 +302,7 @@ class PredictionProcess:
         confidence: int,
         last_timestamp: str,
         comment: str,
+        document: dict[str, Any] | None = None,
     ) -> bool:
         """
         Handle path decisions based on filter results and confidence levels.
@@ -280,6 +328,13 @@ class PredictionProcess:
             confidence: Confidence level from filter validation
             last_timestamp: Last processed timestamp
             comment: Additional information about the filter result
+            document: Metadata describing the filter(s) that produced `path_flag`, as returned by
+                the gate (`None` when nothing was filtered, or when the gate that produced this
+                `path_flag` doesn't build one yet). Written via `write_prediction_metadata` for
+                `CONTINUE` and `REPEAT` alike — both persist a `predictions` row (`REPEAT` reuses
+                the previous row's values but still inserts a new one), so both can reference the
+                resulting `metadata_id`; `STOP` never reaches a write at all, so its `document` is
+                simply never used here.
 
         Returns:
             bool: True if processing should stop, False to continue
@@ -304,7 +359,10 @@ class PredictionProcess:
             # Stop processing and exit workflow
             return True
         elif path_flag == 'REPEAT':
-            # Repeat last prediction if available
+            metadata_id = await self._write_prediction_metadata(metadata, document)
+            # Repeat last prediction, copying the previous row's values forward with a new
+            # timestamp. `repeat_last_prediction` now lives in `Storage` (laborious), not the
+            # shared library — its INSERT...SELECT references `metadata_id` directly.
             await workflow.execute_activity_method(
                 Activities.repeat_last_prediction,
                 {
@@ -313,12 +371,14 @@ class PredictionProcess:
                     'table_name': table_name,
                     'model': model_id,
                     'last_timestamp': last_timestamp,
+                    'metadata_id': metadata_id,
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(minutes=1),
             )
             return True
         elif path_flag == 'CONTINUE':
+            metadata_id = await self._write_prediction_metadata(metadata, document)
             # call write workflow
             await workflow.execute_child_workflow(
                 'subworkflow.format_and_export_prediction',
@@ -335,6 +395,7 @@ class PredictionProcess:
                     'table_name': table_name,
                     'transform_table_name': transform_table_name,
                     'comment': comment,
+                    'metadata_id': metadata_id,
                     'opc_output_config': input_data['opc_output_config'],
                     'pi_web_api_output_config': input_data['pi_web_api_output_config'],
                     'prediction_store_policy': input_data['prediction_store_policy'],

model_convert.ipynb

@@ -1,106 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "code",
-   "execution_count": 23,
-   "id": "e838ff21",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import csv\n",
-    "\n",
-    "def csv_to_tag_lists(csv_path: str) -> dict:\n",
-    "    read_tags = []\n",
-    "    write_tags = []\n",
-    "\n",
-    "    def to_float(val):\n",
-    "        try:\n",
-    "            return float(str(val).strip())\n",
-    "        except Exception:\n",
-    "            return None\n",
-    "\n",
-    "    with open(csv_path, newline=\"\", encoding=\"utf-8\") as f:\n",
-    "        reader = csv.DictReader(f)\n",
-    "        for row in reader:\n",
-    "            # Basic normalization\n",
-    "            op = (row.get(\"operation\") or \"\").strip()\n",
-    "\n",
-    "            if op == \"READ\":\n",
-    "                # Build common tag payload with required mappings\n",
-    "                tag = {\n",
-    "                    \"server_id\": \"1\",\n",
-    "                    \"tag_address\": row.get(\"opc_tag\"),\n",
-    "                    \"tag_name\": row.get(\"name\"),\n",
-    "                    \"data_range\": [to_float(row.get(\"min_value\")), to_float(row.get(\"max_value\"))],\n",
-    "                    \"aggr_func\": row.get(\"aggregation_func\").lower(),\n",
-    "                    # keep other fields with their original names\n",
-    "                    \"frequency\": row.get(\"frequency\"),\n",
-    "                    \"local\": row.get(\"local\"),\n",
-    "                    \"area\": row.get(\"area\"),\n",
-    "                    \"description\": row.get(\"description\"),\n",
-    "                }\n",
-    "\n",
-    "                read_tags.append(tag)\n",
-    "\n",
-    "            else:\n",
-    "                tag = {\n",
-    "                    \"server_id\": \"1\",\n",
-    "                    \"addr\": row.get(\"opc_tag\"),\n",
-    "                    \"tag_name\": row.get(\"name\"),\n",
-    "                    \"local\": row.get(\"local\"),\n",
-    "                    \"area\": row.get(\"area\"),\n",
-    "                    \"description\": row.get(\"description\"),\n",
-    "                }\n",
-    "                \n",
-    "            if op == \"WRITE_PREDICTION\":\n",
-    "                tag[\"type\"] = \"prediction\"\n",
-    "                write_tags.append(tag)\n",
-    "            elif op == \"WRITE_CONFIDENCE\":\n",
-    "                tag[\"type\"] = \"confidence\"\n",
-    "                write_tags.append(tag)\n",
-    "            # ignore any other operation values silently\n",
-    "\n",
-    "    return {\"read_tags\": read_tags, \"write_tags\": write_tags}"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "4621cd43",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import json\n",
-    "\n",
-    "file_names = [\"Courier - Página1.csv\"]\n",
-    "\n",
-    "for file_name in file_names:\n",
-    "    write_file = file_name.replace(\".csv\", \".json\")\n",
-    "\n",
-    "    with open(write_file, \"w\", encoding=\"utf-8\") as f:\n",
-    "        json.dump(csv_to_tag_lists(file_name), f, indent=2, ensure_ascii=False)"
-   ]
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "venv",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.11.13"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "laborious"
-version = "0.0.0"
+version = "0.9.0-rc2"
 description = "Sientia DataOps Laborious - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -60,6 +60,16 @@ ignore = [
     "S105",  # hardcoded passwords ok in tests
     "S106",  # hardcoded passwords ok in tests
 ]
+"e2e/conftest.py" = [
+    # Imports below the sientia/ModelAnalysis stub are load-order critical: the stub
+    # must be in sys.modules before Activities is imported, so they cannot move up.
+    "E402",
+]
+"scripts/**/*.py" = [
+    # Notebook-style scripts split into `# %%` cells: each cell imports what it needs
+    # where it needs it, so imports legitimately appear below other statements.
+    "E402",
+]
 
 [tool.ruff.lint.mccabe]
 max-complexity = 15
@@ -116,12 +126,29 @@ python_functions = ["test_*"]
 addopts = [
     "-v",
     "--strict-markers",
+    # pytest>=9.1 has a known bug where its unraisableexception plugin crashes
+    # (tracemalloc partially-initialized AttributeError) when 2+ unraisable
+    # exceptions land close together — e.g. "coroutine was never awaited" from
+    # AsyncMock-mocked sync methods (metrics_controller, minio_repository) being
+    # GC'd. Harmless mock artifacts turned into a hard ERROR by the plugin itself.
+    "-p", "no:unraisableexception",
+]
+filterwarnings = [
+    # `kafka-python`, pulled in transitively by `sientia_do`, reads its protocol schemas through the
+    # legacy `importlib.resources` API that Python 3.11 deprecated. It emits ~109 warnings per run,
+    # twice each (once at the call site, once inside `importlib`), which is enough noise to bury a
+    # warning worth reading. Nothing here can fix it — the change belongs in `kafka-python`.
+    #
+    # Matched on the message rather than by module so the filter cannot quietly swallow an unrelated
+    # deprecation: any *other* DeprecationWarning, including one from our own code, still shows up.
+    "ignore:read_text is deprecated:DeprecationWarning",
+    "ignore:open_text is deprecated:DeprecationWarning",
 ]
 markers = [
     "asyncio: marks tests as async",
     "integration: marks tests as integration tests",
     "unit: marks tests as unit tests",
-    "opc: marks E2E tests that use in-process asyncua + real OpcRepository",
+    "opc: marks tests that use the in-process OPC UA server (OpcRepository E2E)",
 ]
 
 [tool.coverage.run]
@@ -148,6 +175,7 @@ exclude_lines = [
     "if TYPE_CHECKING:",
     "class .*\\bProtocol\\):",
     "@(abc\\.)?abstractmethod",
+    "@overload",
 ]
 
 [tool.coverage.html]

requirements-dev.txt

@@ -13,7 +13,10 @@ types-requests>=2.31.0   # Type stubs for requests
 pytest>=7.4.0            # Testing framework
 pytest-cov>=4.1.0        # Coverage plugin for pytest
 pytest-asyncio>=0.21.0   # Async test support (already in main requirements)
-testcontainers[postgres,minio] # PostgreSQL and MinIO containers for E2E tests
+testcontainers[postgres,minio,mongodb]>=4.15.0 # PostgreSQL, MinIO and MongoDB containers for E2E tests
+                         # Floor is the `testcontainers.community.*` namespace: the flat
+                         # `testcontainers.minio` / `.postgres` / `.mongodb` paths the e2e suite used
+                         # to import are deprecated shims that warn on import.
 
 # Development Tools
 ipython>=8.12.0          # Enhanced Python shell

requirements-local.txt

@@ -1,18 +1,25 @@
 temporalio
 psycopg2-binary
-sqlalchemy
+# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do
+# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
+sqlalchemy<2.1
 asyncua==1.0.6
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
 git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
 prometheus-client   
-botocore
-boto3
-s3fs
+# fsspec/s3fs pinned as a pair: s3fs pins fsspec exactly, so a free `s3fs` can make the
+# resolver fail against evidently's fsspec>=2024.2.0. Versions match the
+# library-distribution-controller freeze.
+fsspec==2026.7.0
+s3fs==2026.7.0
+# aiobotocore 3.9.0 (pulled by s3fs) requires botocore<1.43.57,>=1.43.3.
+botocore==1.43.56
+boto3==1.43.56
 pyarrow
 kaleido
 hyperopt
 shap
 pycurl
 scipy<1.14.0
-scikit-learn==1.5.2
+scikit-learn==1.5.2

requirements.txt

@@ -1,18 +1,25 @@
 temporalio
 psycopg2-binary
-sqlalchemy
+# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do
+# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
+sqlalchemy<2.1
 asyncua==1.0.6
 redis
-sientia_do>=1.12.1
+sientia_do>=1.12.2
 sientia_model>=0.8.2
 prometheus-client
-botocore
-boto3
-s3fs
+# fsspec/s3fs pinned as a pair: s3fs pins fsspec exactly, so a free `s3fs` can make the
+# resolver fail against evidently's fsspec>=2024.2.0. Versions match the
+# library-distribution-controller freeze.
+fsspec==2026.7.0
+s3fs==2026.7.0
+# aiobotocore 3.9.0 (pulled by s3fs) requires botocore<1.43.57,>=1.43.3.
+botocore==1.43.56
+boto3==1.43.56
 pyarrow
 kaleido
 hyperopt
 shap
 pycurl
 scipy<1.14.0
-scikit-learn==1.5.2
+scikit-learn

sonar-project.properties

@@ -1,4 +1,4 @@
-sonar.projectKey=Aignosi_sientia-dataops-laborious_temporal_beaec423-6c42-4f26-8134-b676287b499d
+sonar.projectKey=Aignosi_sientia-dataops-laborious_temporal_ca1a7039-6db9-49e5-be78-54d29bc93e4f
 sonar.projectName=sientia-dataops-laborious_temporal
 sonar.sources=laborious
 sonar.tests=tests
@@ -8,4 +8,4 @@ sonar.qualitygate.wait=true
 sonar.qualitygate.timeout=300
 sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.xunit.reportPath=pytest.xml
-sonar.python.version=3.11
+sonar.python.version=3.11

tests/laborious/activities/test_activities.py

@@ -6,6 +6,7 @@ from laborious.activities.gates import Gates
 from laborious.activities.mlflow import MLFlow
 from laborious.activities.model_metrics import ModelMetrics
 from laborious.activities.opc import OPC
+from laborious.activities.prediction_metadata import PredictionMetadata
 from laborious.activities.storage import Storage
 
 
@@ -13,6 +14,7 @@ from laborious.activities.storage import Storage
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -22,6 +24,7 @@ def test___init__(
     mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_prediction_metadata_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -47,6 +50,7 @@ def test___init__(
     }
 
     mlflow_repository = MagicMock()
+    mongo_repository = MagicMock()
     plugin_store = MagicMock()
 
     opc_config = {
@@ -73,6 +77,7 @@ def test___init__(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=mongo_repository,
     )
 
     assert isinstance(activities, Activities)
@@ -82,6 +87,7 @@ def test___init__(
     assert isinstance(activities, Gates)
     assert isinstance(activities, ModelMetrics)
     assert isinstance(activities, API)
+    assert isinstance(activities, PredictionMetadata)
 
     mock_storage_init.assert_called_once_with(
         ANY,
@@ -125,6 +131,14 @@ def test___init__(
         metrics_controller=mock_metrics_controller.return_value,
     )
 
+    mock_prediction_metadata_init.assert_called_once_with(
+        ANY,
+        mongo_repository=mongo_repository,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+
     mock_model_metrics_init.assert_called_once_with(
         ANY,
         logger=logger,
@@ -158,6 +172,7 @@ def test___init__(
 @patch('laborious.activities.activities.MLFlow')
 @patch('laborious.activities.activities.OPC')
 @patch('laborious.activities.activities.Gates')
+@patch('laborious.activities.activities.PredictionMetadata')
 @patch('laborious.activities.activities.ModelMetrics')
 @patch('laborious.activities.activities.API')
 @patch('laborious.activities.activities.MinioRepository')
@@ -165,6 +180,7 @@ def test_shutdown(
     _mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_prediction_metadata_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -217,6 +233,7 @@ def test_shutdown(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=MagicMock(),
     )
 
     activities.shutdown()
@@ -226,6 +243,7 @@ def test_shutdown(
     mock_gates_init.close.assert_called_once()
     mock_model_metrics_init.close.assert_called_once()
     mock_api_init.close.assert_called_once()
+    mock_prediction_metadata_init.close.assert_called_once()
 
 
 @patch('laborious.activities.activities.SientiaMLflowRepository')
@@ -234,6 +252,7 @@ def test_shutdown(
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -243,6 +262,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
     mock_minio_repository,
     _mock_api_init,
     _mock_model_metrics_init,
+    _mock_prediction_metadata_init,
     _mock_gates_init,
     _mock_opc_init,
     _mock_mlflow_init,
@@ -286,6 +306,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
         pi_web_api_config=pi_web_api_config,
         logger=logger,
         notification_handler=notification_handler,
+        mongo_repository=MagicMock(),
     )
 
     mock_build_mlflow_config.assert_called_once()
@@ -297,3 +318,92 @@ def test___init___builds_mlflow_repository_when_not_provided(
         notification_handler=notification_handler,
         metrics_controller=mock_metrics_controller.return_value,
     )
+
+
+@patch('laborious.activities.activities.MongoDBRepository')
+@patch('laborious.activities.activities.build_mongodb_config')
+@patch('laborious.activities.activities.Storage.__init__')
+@patch('laborious.activities.activities.MLFlow.__init__')
+@patch('laborious.activities.activities.OPC.__init__')
+@patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.PredictionMetadata.__init__')
+@patch('laborious.activities.activities.ModelMetrics.__init__')
+@patch('laborious.activities.activities.API.__init__')
+@patch('laborious.activities.activities.MinioRepository')
+@patch('laborious.activities.activities.MetricsController')
+def test___init___builds_mongo_repository_when_not_provided(
+    mock_metrics_controller,
+    mock_minio_repository,
+    _mock_api_init,
+    _mock_model_metrics_init,
+    mock_prediction_metadata_init,
+    mock_gates_init,
+    _mock_opc_init,
+    _mock_mlflow_init,
+    _mock_storage_init,
+    mock_build_mongodb_config,
+    mock_mongo_repository_cls,
+):
+    postgres_config = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10,
+    }
+    minio_config = {
+        'endpoint_url': 'localhost:9000',
+        'access_key': 'minio',
+        'secret_key': 'minio123',
+        'default_bucket': 'test',
+        'retention_hours': 24,
+        'secure': False,
+    }
+    opc_config = {'bootstrap_servers': 'localhost:9092', 'polling_time': 1000, 'group_id': 'test'}
+    pi_web_api_config = {'base_url': 'https://pi', 'auth_type': 'bearer', 'auth_token': 'token'}
+    logger = MagicMock()
+    notification_handler = MagicMock()
+    plugin_store = MagicMock()
+    mlflow_repository = MagicMock()
+    mock_build_mongodb_config.return_value = {
+        'connection_string': 'mongodb://root:pass@localhost:27018',
+        'database_name': 'sientia',
+    }
+
+    Activities(
+        postgres_config=postgres_config,
+        plugin_store=plugin_store,
+        minio_config=minio_config,
+        opc_config=opc_config,
+        pi_web_api_config=pi_web_api_config,
+        logger=logger,
+        notification_handler=notification_handler,
+        mlflow_repository=mlflow_repository,
+    )
+
+    mock_build_mongodb_config.assert_called_once()
+    mock_mongo_repository_cls.assert_called_once_with(
+        connection_string='mongodb://root:pass@localhost:27018',
+        database_name='sientia',
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    mock_gates_init.assert_called_once_with(
+        ANY,
+        minio_repository=mock_minio_repository.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    # The same repository built here is the one handed to PredictionMetadata, not a second
+    # instance.
+    mock_prediction_metadata_init.assert_called_once_with(
+        ANY,
+        mongo_repository=mock_mongo_repository_cls.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )

tests/laborious/activities/test_api.py

@@ -2,6 +2,7 @@ from unittest.mock import ANY, MagicMock, call, patch
 
 from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
 from laborious.activities.api import API, PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
 
@@ -70,7 +71,7 @@ def test_get_pi_web_api_core_labels_without_operation_type(mock_pi_web_api_clien
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     with patch.object(
         SientiaMonitoring,
@@ -104,7 +105,7 @@ def test_get_pi_web_api_core_labels_with_operation_type(mock_pi_web_api_client):
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     with patch.object(
         SientiaMonitoring,
@@ -131,7 +132,7 @@ def test__init__():
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
     assert api.pi_web_api_client is not None
@@ -152,7 +153,7 @@ def api(mock_pi_web_api_client):
         auth_token='test_token',
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     api_instance.send_notification = MagicMock()
     api_instance.info = MagicMock()
@@ -218,6 +219,43 @@ def test_write_pi_web_api_data_success(mock_dataframe, api, base_input_data):
     }
 
 
+@patch('laborious.activities.api.DataFrame')
+def test_write_pi_web_api_data_partial_prediction_mismatch_downgrades_confidence(
+    mock_dataframe, api, base_input_data
+):
+    """A partial write (some tags report Errors) doesn't raise — it's caught by the tag-count
+    mismatch check, which still must downgrade confidence and attach the mismatch message."""
+    input_data = {
+        **base_input_data,
+        'pi_web_api_output_config': {
+            'endpoint': 'https://test-pi-server.com/piwebapi',
+            'prediction_tags': {'tag1': 'web_id_1', 'tag2': 'web_id_2'},
+            'confidence_tags': {'tag3': 'web_id_3'},
+        },
+    }
+    mock_df = _create_mock_dataframe()
+    mock_dataframe.return_value = mock_df
+
+    api.pi_web_api_client.write_value.side_effect = [
+        [{'WebId': 'web_id_1', 'Errors': []}, {'WebId': 'web_id_2', 'Errors': ['bad tag']}],
+        [{'WebId': 'web_id_3', 'Errors': []}],
+    ]
+
+    api.write_pi_web_api_data(input_data)
+
+    mock_df.__setitem__.assert_any_call(
+        'prediction_confidence', PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
+    )
+    mock_df.__setitem__.assert_any_call('comments', ANY)
+    api.send_notification.assert_called_once_with(
+        metadata=metadata['metadata'],
+        notification_id='WRITE_PI_WEB_API_PREDICTION_ERROR',
+        message=ANY,
+        block='write_pi_web_api_data',
+        level=NotificationLevel.ERROR,
+    )
+
+
 @patch('laborious.activities.api.DataFrame')
 def test_write_pi_web_api_data_prediction_error(mock_dataframe, api, base_input_data):
     mock_dataframe.return_value = _create_mock_dataframe(
@@ -348,7 +386,8 @@ def test_close(mock_shutdown, api):
     api.close()
 
     api.pi_web_api_client.close.assert_called_once()
-    mock_shutdown.assert_called_once_with(api)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(api)) == 1
 
 
 def test_process_pi_web_api_response_success(api):

tests/laborious/activities/test_gates.py

@@ -3,9 +3,10 @@ from unittest.mock import ANY, MagicMock, call, patch
 from pandas import DataFrame
 from pytest import fixture
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 
-from laborious.activities.gates import Gates
+from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions
 
 
 @fixture(autouse=True)
@@ -38,7 +39,7 @@ def gates_activity():
     gates = Gates(
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.error = MagicMock()
     gates.debug = MagicMock()
@@ -60,6 +61,13 @@ metadata = {
 }
 
 
+def test_input_filter_counters_covers_every_input_filter_function():
+    # A filter added to input_filter_functions without a matching entry here would degrade to
+    # counter=0 (see the `.get(..., lambda *_: 0)` fallback in input_gate) instead of crashing —
+    # but this test should still catch the gap before it ships (review finding, 2026-09-30).
+    assert set(input_filter_counters) == set(input_filter_functions)
+
+
 def test_input_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {
@@ -73,7 +81,7 @@ def test_input_gate_invalid_filter(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.error.assert_called_once_with(
         'Filter INVALID_FILTER not found', metadata['metadata']
     )
@@ -97,7 +105,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.send_notification.assert_called_once_with(
         metadata=metadata['metadata'],
         notification_id='INTPUT_GATE_ERROR__EMPTY_DATA',
@@ -121,7 +129,7 @@ def test_input_gate_no_filters(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -135,10 +143,14 @@ def test_input_gate_with_filter(gates_activity):
     }
 
     # Act
-    result = gates_activity.input_gate(input_data)
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
 
-    # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    # Assert: the gate only computes the document now — it never writes anywhere, and never
+    # checks path_flag to decide whether to build one (that decision moved to the caller, see
+    # `PredictionMetadata.write_prediction_metadata` and `prediction_process.py`), so even a STOP
+    # resolution still gets its document built.
+    assert (path_flag, confidence, comment) == ('STOP', -1, 'Input data with bad quality')
+    assert document == {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
     gates_activity.debug.assert_called()
 
 
@@ -155,7 +167,12 @@ def test_input_gate_with_filter_lowercase_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_capitalized_keys(gates_activity):
@@ -171,7 +188,12 @@ def test_input_gate_with_filter_capitalized_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_not_caught(gates_activity):
@@ -187,10 +209,94 @@ def test_input_gate_with_filter_not_caught(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
+def test_input_gate_repeat_also_builds_document(gates_activity):
+    # Arrange: REPEAT used to be treated like STOP (no document at all) — it isn't anymore, since
+    # REPEAT does insert a new `predictions` row (reusing the previous row's values), so it needs
+    # a document/metadata_id just like CONTINUE (2026-10-01 redesign).
+    input_data = {
+        **metadata,
+        'filters': {'EMPTY_DATA': {'POLICY': 'REPEAT', 'CONFIG': {}}},
+        'data': _minio_payload(DataFrame({'value': []})),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    result = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert result == (
+        'REPEAT',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]},
+    )
+
+
+def test_input_gate_continue_builds_document_with_counter(gates_activity):
+    # Arrange: two rows with a null in one of the configured variables ('x'); a third row is
+    # clean, to confirm the counter reflects only the rows that actually matched the condition.
+    input_data = {
+        **metadata,
+        'filters': {
+            'SPECIFIC_VARIABLES_NULL_VALUES': {
+                'POLICY': 'CONTINUE',
+                'CONFIG': {'variables': ['x']},
+            }
+        },
+        'data': _minio_payload(
+            DataFrame({'variable': ['x', 'x', 'y'], 'value': [None, None, None]})
+        ),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert (path_flag, confidence, comment) == ('CONTINUE', 2, 'Input data with bad quality')
+    assert document == {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
+
+
+def test_input_gate_multiple_filters_same_policy_lists_each(gates_activity):
+    # Arrange: force both input filters to trigger (not reachable with today's real filter
+    # conditions — EMPTY_DATA and SPECIFIC_VARIABLES_NULL_VALUES are mutually exclusive) to lock
+    # in the list-building behavior approved by Vitor (2026-09-30): one entry per filter sharing
+    # the winning policy, instead of a single joined/aggregated value.
+    with patch('laborious.activities.gates.input_filter_functions') as mock_filters:
+        mock_filters.__contains__.return_value = True
+        mock_filters.__getitem__.return_value = MagicMock(return_value=True)
+
+        input_data = {
+            **metadata,
+            'filters': {
+                'SPECIFIC_VARIABLES_NULL_VALUES': {
+                    'POLICY': 'CONTINUE',
+                    'CONFIG': {'variables': ['x']},
+                },
+                'EMPTY_DATA': {'POLICY': 'CONTINUE', 'CONFIG': {}},
+            },
+            'data': _minio_payload(DataFrame({'variable': ['x'], 'value': [None]})),
+            'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+        }
+
+        # Act
+        path_flag, _, _, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert path_flag == 'CONTINUE'
+    assert {f['name'] for f in document['filters']} == {
+        'SPECIFIC_VARIABLES_NULL_VALUES',
+        'EMPTY_DATA',
+    }
+    assert all(f['policy'] == 'CONTINUE' for f in document['filters'])
+
+
 def test_mlflow_response_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {
@@ -985,12 +1091,13 @@ def test_close_disposes_minio_repository(mock_shutdown):
         minio_repository=minio,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.close()
     minio.close.assert_called_once()
     assert gates.minio_repository is None
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1
 
 
 @patch.object(SientiaMonitoring, 'shutdown')
@@ -1000,7 +1107,8 @@ def test_close_without_minio_repository(mock_shutdown):
         minio_repository=None,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.close()
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1

tests/laborious/activities/test_mlflow.py

@@ -5,6 +5,7 @@ import numpy as np
 import pandas as pd
 from pytest import fixture, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 
 from laborious.activities.mlflow import MLFlow
@@ -22,7 +23,7 @@ def _passthrough_from_dict():
 def test___init__(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     mlflow_repo = MagicMock()
     plugin_store = MagicMock()
 
@@ -63,7 +64,7 @@ def test___init__(mock_minio_repository):
 def mlflow(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     mlflow_repo = MagicMock()
     plugin_store = MagicMock()
 
@@ -374,9 +375,9 @@ def test_request_predict_failure(mock_to_datetime, mock_from_dataframe, mlflow):
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_data_success_retrain(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
 
     mv_alias = MagicMock()
@@ -427,9 +428,9 @@ def test_retrain_model_success_data_success_retrain(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_with_payload_data(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -472,9 +473,9 @@ def test_retrain_model_success_with_payload_data(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -854,3 +855,22 @@ def test_get_reference_data_exception(mlflow):
     result = mlflow.get_reference_data(input_data)
 
     assert result is None
+
+
+def test_find_downloaded_csv_prefers_the_direct_path(mlflow, tmp_path):
+    direct = tmp_path / 'evaluation_data.csv'
+    direct.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == direct
+
+
+def test_find_downloaded_csv_searches_nested_directories(mlflow, tmp_path):
+    nested = tmp_path / 'artifacts' / 'data' / 'evaluation_data.csv'
+    nested.parent.mkdir(parents=True)
+    nested.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == nested
+
+
+def test_find_downloaded_csv_returns_none_when_missing(mlflow, tmp_path):
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') is None

tests/laborious/activities/test_model_metrics.py

@@ -1,8 +1,9 @@
 from unittest.mock import ANY, MagicMock, patch
 
-from pandas import DataFrame, Timestamp
-from pytest import fixture, raises
+from pandas import DataFrame, Series, Timestamp
+from pytest import fixture, mark, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 from sientia_model.analytics.drift_analysis import DriftInsufficientDataError
 
@@ -14,7 +15,7 @@ def model_metrics_activity():
     model_metrics = ModelMetrics(
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     model_metrics.error = MagicMock()
     model_metrics.debug = MagicMock()
@@ -952,3 +953,50 @@ def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity)
 
     assert len(result['metric']) == 1
     assert result['metric'].values[0] == 'rmse'
+
+
+@mark.parametrize(
+    'failing_stage',
+    ['detect_univariate_drift', 'detect_multivariate_drift', 'get_drift_metrics_dataframe'],
+)
+@patch('laborious.activities.model_metrics.DriftAnalysis')
+def test_get_drift_metrics_reraises_insufficient_data_untouched(
+    mock_drift_analysis, failing_stage, model_metrics_activity
+):
+    """Insufficient data is not a stage failure: it propagates without the stage error report."""
+    error = DriftInsufficientDataError('not enough rows', analysis_rows=1)
+    getattr(mock_drift_analysis.return_value, failing_stage).side_effect = error
+    data = DataFrame({'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]})
+    model_metrics_activity._drift_analyze_stage_error = MagicMock()
+
+    with raises(DriftInsufficientDataError) as exc_info:
+        model_metrics_activity.get_drift_metrics(
+            reference_data=data,
+            target_data=data,
+            target_name='target',
+            reference_columns=data.drop(columns=['target', 'timestamp']).columns,
+            drift_metrics=['ks_test'],
+            chunk_period='min',
+            metadata=metadata['metadata'],
+        )
+
+    assert exc_info.value is error
+    model_metrics_activity._drift_analyze_stage_error.assert_not_called()
+
+
+def test_to_naive_utc_converts_aware_timestamps_to_naive_utc():
+    series = Series(['2024-01-01T03:00:00-03:00', '2024-01-01T09:00:00-03:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00'), Timestamp('2024-01-01 12:00:00')]
+
+
+def test_to_naive_utc_keeps_naive_timestamps():
+    series = Series(['2024-01-01 06:00:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00')]

tests/laborious/activities/test_opc.py

@@ -4,6 +4,7 @@ import pytest
 from pandas import DataFrame
 from pytest import mark
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
 from laborious.activities.opc import (
     OPC,
@@ -32,7 +33,7 @@ def test__init__():
         opc_servers=servers,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
     assert opc.opc_servers == servers
@@ -43,7 +44,7 @@ def test__init__():
 @patch('laborious.activities.opc.OPC.send_notification')
 def test_init_opc(mock_send_notification, mock_opc_repository):
     mock_logger = MagicMock()
-    mock_metrics_controller = MagicMock()
+    mock_metrics_controller = MagicMock(spec=MetricsController)
     server1 = MagicMock(
         connect=MagicMock(return_value=(True, {})), write_data=MagicMock(return_value=(True, {}))
     )
@@ -192,7 +193,7 @@ def opc(mock_opc_repository):
         opc_servers=servers,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     opc.init_opc()
     opc.send_notification = MagicMock()
@@ -539,6 +540,38 @@ def test_manage_output_tags_do_nothing(opc):
     opc._write_tags_from_config.assert_not_called()
 
 
+@patch('laborious.activities.opc.DataFrame')
+def test_write_opc_data_propagates_session_bad_and_reconnect_in_progress(mock_dataframe, opc):
+    """A single server reporting session_bad/reconnect_in_progress must flip the aggregated
+    flags passed to process_confidence, even though nothing raised."""
+    input_data = {
+        **metadata,
+        'data': {'prediction': [0.75], 'prediction_confidence': [0.95]},
+        'opc_output_config': {
+            'server1': {
+                'prediction_tags': {'tag1': {'data_type': 'float'}},
+                'confidence_tags': {'tag2': {'data_type': 'float'}},
+            }
+        },
+    }
+
+    opc.manage_output_tags = MagicMock(
+        return_value=(False, {'tag1': 0.1, 'tag2': 0.2}, True, 'BadSessionClosed', True)
+    )
+    opc.process_confidence = MagicMock(return_value={'data': 'data'})
+
+    opc.write_opc_data(input_data)
+
+    opc.process_confidence.assert_called_once_with(
+        mock_dataframe.return_value,
+        False,
+        metadata['metadata'],
+        session_bad=True,
+        opc_status='BadSessionClosed',
+        reconnect_in_progress=True,
+    )
+
+
 @patch('laborious.activities.opc.DataFrame')
 def test_write_opc_data_success(mock_dataframe, opc):
     # Arrange

tests/laborious/activities/test_prediction_metadata.py

@@ -0,0 +1,179 @@
+from unittest.mock import MagicMock, call, patch
+
+from pytest import fixture
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+from laborious.activities.prediction_metadata import PredictionMetadata
+
+
+@fixture
+def mongo() -> MagicMock:
+    return MagicMock()
+
+
+@fixture
+def prediction_metadata(mongo: MagicMock):
+    activity = PredictionMetadata(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    return activity
+
+
+metadata = {
+    'metadata': {
+        'model_id': 'test_model',
+        'model_name': 'test_model',
+        'workflow_name': 'test_workflow',
+        'schema_name': 'test_schedule',
+    },
+}
+
+
+def test_write_prediction_metadata_no_documents_skips_mongo(prediction_metadata, mongo: MagicMock):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': []})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_missing_documents_key_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_documents_with_no_filters_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: a falsy document (None) and one with an empty `filters` list — neither contributes
+    # anything to write.
+    documents = [None, {'filters': []}]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_not_configured_returns_none():
+    # Arrange
+    activity = PredictionMetadata(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = activity.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert
+    assert result is None
+    activity.error.assert_not_called()
+
+
+def test_write_prediction_metadata_success_generates_id_and_inserts(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)),
+    # not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side.
+    assert result is not None
+    assert len(result) == 24
+    int(result, 16)  # raises ValueError if not hex
+
+    collection, inserted_document, inserted_metadata = mongo.insert.call_args.args
+    assert collection == PredictionMetadata._PREDICTION_METADATA_COLLECTION
+    assert inserted_document == {'filters': document['filters'], '_id': result}
+    assert inserted_metadata == metadata['metadata']
+
+
+def test_write_prediction_metadata_combines_multiple_documents(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: two documents (e.g. from two different sources contributing metadata for the same
+    # prediction cycle) — their `filters` entries are flattened into one combined document.
+    documents = [
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]},
+        {
+            'filters': [
+                {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}
+            ]
+        },
+    ]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is not None
+    _, inserted_document, _ = mongo.insert.call_args.args
+    assert inserted_document['filters'] == [
+        {'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0},
+        {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2},
+    ]
+
+
+def test_write_prediction_metadata_insert_failure_swallowed(prediction_metadata, mongo: MagicMock):
+    # Arrange
+    mongo.insert.side_effect = RuntimeError('connection reset')
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert: the prediction write must never be blocked or failed by this
+    assert result is None
+    prediction_metadata.error.assert_called_once()
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_disposes_mongo_repository(mock_shutdown):
+    """``PredictionMetadata.close`` should close the Mongo client and clear the reference."""
+    mongo = MagicMock()
+    activity = PredictionMetadata(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    mongo.close.assert_called_once()
+    assert activity.mongo_repository is None
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_without_mongo_repository(mock_shutdown):
+    """When no Mongo repository is configured, ``close`` only shuts down monitoring."""
+    activity = PredictionMetadata(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1

tests/laborious/activities/test_storage.py

@@ -1,9 +1,10 @@
 import datetime
 import os
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 from pytest import fixture, raises
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 from sientia_do.temporal.activities.postgres_sync import Postgres
 
@@ -52,7 +53,7 @@ def storage(mock_minio_repository):
         minio_repository=mock_minio_repository.return_value,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
 
 
@@ -60,7 +61,7 @@ def storage(mock_minio_repository):
 def test___init___not_hasattr(mock_minio_repository):
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     minio_repo = mock_minio_repository.return_value
     storage = Storage(
         host='localhost',
@@ -87,7 +88,7 @@ def test___init___none_minio_repository(mock_minio_repository, storage):
     storage.minio_repository = None
     logger = MagicMock()
     notification_handler = MagicMock()
-    metrics_controller = MagicMock()
+    metrics_controller = MagicMock(spec=MetricsController)
     storage.__init__(
         host='localhost',
         port=5432,
@@ -121,7 +122,7 @@ def test___init___done_repository(mock_minio_repository, storage):
         minio_repository=mock_minio_repository.return_value,
         logger=MagicMock(),
         notification_handler=MagicMock(),
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     mock_minio_repository.assert_not_called()
     assert storage.minio_repository is not None
@@ -133,7 +134,8 @@ def test_close(storage, _patch_monitoring_shutdown):
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shutdown):
@@ -143,7 +145,8 @@ def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shu
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_load_query_with_minio_offload_no_rows(storage):
@@ -254,6 +257,86 @@ def test_export_payload_to_postgres(storage):
     assert result == {'success': True}
 
 
+def test_build_repeat_last_prediction_query(storage):
+    query = storage._build_repeat_last_prediction_query('public', 'predictions')
+    query_str = str(query)
+
+    assert 'INSERT INTO "public".predictions' in query_str
+    assert 'metadata_id' in query_str
+    assert ':last_timestamp' in query_str
+    assert ':metadata_id' in query_str
+    assert ':model' in query_str
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_success(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    text_mock.assert_called_once()
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': 'abc123'},
+    )
+    storage.observe_lag_sync.assert_called_once()
+    storage.emit_metric_sync.assert_called_once()
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_without_metadata_id(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': None},
+    )
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_error(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.send_notification = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with (
+        patch('sqlalchemy.orm.session.Session.execute', side_effect=ValueError('db')),
+        raises(ValueError),
+    ):
+        storage.repeat_last_prediction(query_items)
+
+    storage.send_notification.assert_called_once()
+    storage.emit_metric_sync.assert_called()
+
+
 def test_cleanup_minio_objects_expired_minio_not_initialized(storage):
     storage.minio_repository = None
 
@@ -322,3 +405,17 @@ def test_cleanup_minio_objects_expired_list_objects_error(mock_now, storage):
         attachment_content=ANY,
     )
     storage.error.assert_called_once()
+
+
+def test_close_without_engine_skips_postgres(storage, _patch_monitoring_shutdown):
+    """A Storage whose Postgres init never ran (no engine) still closes MinIO."""
+    del storage.engine
+    minio = MagicMock()
+    storage.minio_repository = minio
+
+    with patch('laborious.activities.storage.Postgres.close') as mock_postgres_close:
+        storage.close()
+
+    mock_postgres_close.assert_not_called()
+    minio.close.assert_called_once()
+    assert storage.minio_repository is None

tests/laborious/utils/models/test_minio_dataframe_payload.py

@@ -220,6 +220,26 @@ def test_from_dataframe_offloaded(mock_now):
     minio.upload_file.assert_called_once()
 
 
+@patch('laborious.utils.models.minio_dataframe_payload.OFFLOAD_THRESHOLD_BYTES', 10**9)
+def test_from_dataframe_inline_keeps_a_provided_last_timestamp():
+    """An explicit last_timestamp must survive as-is, skipping the max() derivation."""
+    minio = MagicMock()
+    logger = MagicMock()
+    df = _mock_dataframe({'timestamp': ['2024-01-01'], 'value': [42]})
+
+    result = MinioDataFramePayload.from_dataframe(
+        dataframe=df,
+        minio_repo=minio,
+        model_name='m',
+        operation='initial',
+        last_timestamp='2030-01-01T00:00:00+00:00',
+        logger=logger,
+    )
+
+    assert result.last_timestamp == '2030-01-01T00:00:00+00:00'
+    logger.custom_debug.assert_called()
+
+
 def test_from_dict_inline():
     raw = {
         'last_timestamp': '2024-01-01T00:00:00+00:00',

tests/laborious/utils/repository/test_opc_repository.py

@@ -1,5 +1,7 @@
 import concurrent.futures
 import json
+import threading
+import time
 from datetime import datetime
 from unittest.mock import ANY, MagicMock, Mock, patch
 
@@ -7,12 +9,15 @@ import pytest
 from asyncua.crypto import security_policies
 from asyncua.ua.uaerrors import BadNodeIdUnknown, BadSessionIdInvalid
 from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.metrics_controller import MetricsController
 
+from laborious import metrics
 from laborious.utils.repository.opc_repository import (
     OpcClientAlreadyExistsError,
     OpcClientNotInitializedError,
     OpcRepository,
     OpcSessionAlreadyConnectedError,
+    _opc_authentication_token_str,
     is_reconnectable_opcua_bad,
 )
 
@@ -35,7 +40,7 @@ def opc_repository(mock_logger):
         cert_path='/path/to/cert.pem',
         private_key_path='/path/to/key.pem',
         server_cert_path='/path/to/server_cert.pem',
-        metrics_controller=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
     )
     repository.disconnection_interval = 0.1
     repository.send_notification = MagicMock()
@@ -86,6 +91,40 @@ def test_init(opc_repository):
     assert opc_repository.last_reconnection_time is None
 
 
+def test_opc_authentication_token_str_no_protocol():
+    client = MagicMock()
+    client.aio_obj.uaclient.protocol = None
+
+    assert _opc_authentication_token_str(client) == 'unknown'
+
+
+def test_opc_authentication_token_str_no_token():
+    client = MagicMock()
+    client.aio_obj.uaclient.protocol = MagicMock(authentication_token=None)
+
+    assert _opc_authentication_token_str(client) == 'unknown'
+
+
+def test_opc_authentication_token_str_swallows_errors():
+    class _ExplodingClient:
+        @property
+        def aio_obj(self):
+            raise RuntimeError('boom')
+
+    assert _opc_authentication_token_str(_ExplodingClient()) == 'unknown'
+
+
+def test_is_session_open_swallows_errors(opc_repository):
+    class _ExplodingClient:
+        @property
+        def aio_obj(self):
+            raise RuntimeError('boom')
+
+    opc_repository.client = _ExplodingClient()
+
+    assert opc_repository._is_session_open() is False
+
+
 def test_set_security(opc_repository, mock_client):
     opc_repository.client = mock_client
     opc_repository.set_security()
@@ -112,6 +151,13 @@ def test_set_security_missing_certificates(opc_repository):
         assert str(e) == 'Certificate and private key paths must be provided for secure connection.'
 
 
+def test_set_security_missing_server_uri(opc_repository):
+    opc_repository.server_uri = None
+
+    with pytest.raises(ValueError, match='server_uri must be provided for secure connection.'):
+        opc_repository.set_security()
+
+
 def test_set_security_missing_client(opc_repository):
     opc_repository.client = None
     try:
@@ -160,6 +206,43 @@ def test_create_client_raises_when_client_exists(opc_repository, mock_client):
         opc_repository._create_client()
 
 
+def test_create_client_builds_a_secured_client(opc_repository, mock_client):
+    opc_repository.pod_id = 'pod-1'
+    opc_repository.set_security = MagicMock()
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    assert mock_client.aio_obj.name == 'pod-1'
+    assert mock_client.application_uri == 'pod:1'
+    assert mock_client.aio_obj.product_uri == 'pod:1'
+    opc_repository.set_security.assert_called_once()
+
+
+def test_create_client_skips_security_without_a_cert_path(opc_repository, mock_client):
+    opc_repository.cert_path = None
+    opc_repository.set_security = MagicMock()
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    opc_repository.set_security.assert_not_called()
+
+
+def test_connect_locked_reuses_an_existing_client(opc_repository, mock_client):
+    """A client already set (e.g. surviving a prior failed open) is not recreated."""
+    opc_repository.client = mock_client
+    opc_repository._is_session_open = MagicMock(return_value=False)
+    opc_repository._create_client = MagicMock()
+    opc_repository._open_session = MagicMock(return_value=(True, {}))
+
+    result = opc_repository._connect_locked()
+
+    opc_repository._create_client.assert_not_called()
+    opc_repository._open_session.assert_called_once()
+    assert result == (True, {})
+
+
 def test_open_session_success(opc_repository):
     closed_proto = MagicMock()
     closed_proto.state = 'closed'
@@ -361,6 +444,22 @@ def test_write_data_get_node_failed(opc_repository):
     assert error_data['attachment_content'] is not None
 
 
+def test_write_data_without_client_fails_as_get_node_error(opc_repository):
+    """A client torn down between validation and the write is reported, not asserted away."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = None
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    assert is_success is False
+    assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}'
+    assert 'OPC client is not initialized' in error_data['message']
+    last_call = opc_repository.emit_metric_sync.call_args_list[-1]
+    assert last_call.args[1]['result'] == 'GetNodeError:OpcClientNotInitializedError'
+
+
 def test_write_data_invalid_data_type(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
@@ -423,6 +522,75 @@ def test_write_data_write_value_failed(opc_repository, mock_client):
     assert error_data['attachment_content'] is not None
 
 
+def test_write_data_with_empty_metadata_uses_unknown_labels(opc_repository, mock_client):
+    """No model_id/model_name in the write metadata: the metric labels fall back to 'unknown'."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    mock_client.get_node = MagicMock(return_value=MagicMock())
+
+    result = opc_repository.write_data('ns=2;s=TestNode', 42.0, 'float', {})
+
+    assert result == (True, {'response_time': ANY})
+    last_call = opc_repository.emit_metric_sync.call_args_list[-1]
+    assert last_call.args[0] is metrics.OPC_WRITE_ATTEMPTS_TOTAL
+    assert last_call.args[1]['model_id'] == 'unknown'
+    assert last_call.args[1]['model_name'] == 'unknown'
+
+
+def test_write_data_get_node_bad_session_id_schedules_reconnect(opc_repository, mock_client):
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    opc_repository._start_reconnect = MagicMock()
+    mock_client.get_node = MagicMock(side_effect=BadSessionIdInvalid())
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'session_bad'
+    assert error_data['opc_status'] == 'BadSessionIdInvalid'
+
+
+def test_write_data_logs_inter_arrival_and_flags_a_stale_session(opc_repository, mock_client):
+    """The second write on the same session logs the gap since the previous one, and flags it
+    when the gap outlives the session's revised timeout."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = mock_client
+    mock_client.aio_obj.session_timeout = 0.001  # 1 microsecond: any real gap will exceed it
+    mock_client.get_node = MagicMock(return_value=MagicMock())
+
+    opc_repository.write_data('ns=2;s=TestNode', 1.0, 'float', metadata['metadata'])
+    opc_repository.write_data('ns=2;s=TestNode', 2.0, 'float', metadata['metadata'])
+
+    logged = [c.args[0] for c in opc_repository.info.call_args_list]
+    assert any('OPC write inter-arrival' in message for message in logged)
+    emitted_metrics = [c.args[0] for c in opc_repository.emit_metric_sync.call_args_list]
+    assert metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL in emitted_metrics
+
+
+def test_log_write_inter_arrival_skips_the_metric_without_a_client(opc_repository):
+    opc_repository._last_write_mono = time.monotonic() - 1
+    opc_repository.client = None
+
+    opc_repository._log_write_inter_arrival('session-1', 'ns=2;s=Node')
+
+    opc_repository.emit_metric_sync.assert_not_called()
+
+
+def test_log_write_inter_arrival_skips_the_metric_within_the_session_timeout(
+    opc_repository, mock_client
+):
+    opc_repository._last_write_mono = time.monotonic()
+    opc_repository.client = mock_client
+    mock_client.aio_obj.session_timeout = 600_000  # 600s: comfortably longer than this test takes
+
+    opc_repository._log_write_inter_arrival('session-1', 'ns=2;s=Node')
+
+    opc_repository.emit_metric_sync.assert_not_called()
+
+
 def test_is_reconnectable_opcua_bad():
     assert is_reconnectable_opcua_bad(BadSessionIdInvalid()) is True
     assert is_reconnectable_opcua_bad(BadNodeIdUnknown()) is False
@@ -472,6 +640,23 @@ def test_start_reconnect_skips_within_interval(opc_repository):
     assert opc_repository._reconnect_thread is None
 
 
+def test_start_reconnect_skips_when_already_in_progress(opc_repository):
+    opc_repository.last_reconnection_time = None
+    release = threading.Event()
+    in_progress_thread = threading.Thread(target=release.wait, daemon=True)
+    in_progress_thread.start()
+    opc_repository._reconnect_thread = in_progress_thread
+
+    try:
+        opc_repository._start_reconnect('BadSessionIdInvalid', 'tok')
+
+        assert opc_repository._reconnect_thread is in_progress_thread
+        opc_repository.warning.assert_called_once()
+    finally:
+        release.set()
+        in_progress_thread.join(timeout=5.0)
+
+
 def test_write_data_protocol_closed_schedules_reconnect(opc_repository):
     opc_repository.client = MagicMock()
     opc_repository.client.aio_obj.uaclient.protocol = MagicMock(state='closed')
@@ -582,3 +767,77 @@ def test_reconnect_locked_sets_last_reconnection_time(mock_datetime, opc_reposit
     opc_repository._connect_locked.assert_called_once()
     assert result == (True, {})
     assert opc_repository.last_reconnection_time == datetime(2025, 1, 1, 12, 0, 0)
+
+
+def test_run_reconnect_failed_without_error_detail_logs_once(opc_repository):
+    """A falsy `error` (no message to add) must not log a second, empty line."""
+    opc_repository._reconnect_locked = MagicMock(return_value=(False, {}))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    assert opc_repository.error.call_count == 1
+
+
+def test_run_reconnect_swallows_an_unexpected_exception(opc_repository):
+    """A bug inside the reconnect itself must not escape the background thread."""
+    opc_repository._reconnect_locked = MagicMock(side_effect=RuntimeError('boom'))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    assert opc_repository.error.call_count >= 2
+
+
+def test_create_client_without_watchdog_attribute(opc_repository, mock_client):
+    """Older asyncua clients have no `watchdog_intervall`; it is not invented on them."""
+    opc_repository.cert_path = None
+    mock_client.aio_obj = MagicMock(spec=['name', 'description', 'product_uri', 'uaclient'])
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    assert not hasattr(mock_client.aio_obj, 'watchdog_intervall')
+
+
+def test_run_reconnect_success_logs_no_error(opc_repository):
+    opc_repository._reconnect_locked = MagicMock(return_value=(True, {}))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    opc_repository._reconnect_locked.assert_called_once()
+    opc_repository.error.assert_not_called()
+
+
+@pytest.mark.parametrize('failure', [AttributeError('no write_value'), TypeError('bad args')])
+def test_write_node_value_falls_back_to_set_value(failure):
+    node = MagicMock()
+    node.write_value.side_effect = failure
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.write_value.assert_called_once_with('ua-data')
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_node_value_uses_set_value_without_write_value():
+    node = MagicMock(spec=['set_value'])
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_data_session_not_ready_reports_the_reconnect_it_started(opc_repository):
+    """When the SessionNotReady reconnect starts, the write reports reconnect_in_progress."""
+    opc_repository._session_ready.clear()
+    opc_repository._reconnect_thread_in_progress = MagicMock(side_effect=[False, True])
+    opc_repository._start_reconnect = MagicMock()
+    opc_repository.validate_connection = MagicMock()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once_with('SessionNotReady', ANY)
+    opc_repository.validate_connection.assert_not_called()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'reconnect_in_progress'

tests/laborious/utils/test_dataframe_debug.py

@@ -1,12 +1,40 @@
+"""Debug message formatting: full dataframe when small, a summary when it would flood the log."""
+
 from pandas import DataFrame
 
-from laborious.utils.dataframe_debug import build_dataframe_debug_message
+from laborious.utils.dataframe_debug import (
+    DEFAULT_MAX_DEBUG_DATAFRAME_ROWS,
+    build_dataframe_debug_message,
+)
 
 
-def test_build_dataframe_debug_message_skips_large_dataframe():
-    df = DataFrame({'a': [1, 2, 3]})
+def test_non_dataframe_payloads_are_stringified():
+    assert build_dataframe_debug_message('payload:', {'a': 1}) == "payload: {'a': 1}"
 
-    msg = build_dataframe_debug_message('payload', df, max_rows=1)
 
-    assert 'skipped because dataframe has 3 rows' in msg
-    assert '(max: 1)' in msg
+def test_a_small_dataframe_is_logged_in_full():
+    data = DataFrame({'a': [1, 2]})
+
+    message = build_dataframe_debug_message('payload:', data)
+
+    assert message.startswith('payload:\n')
+    assert data.to_csv() in message
+
+
+def test_a_dataframe_over_the_row_limit_is_summarized_instead():
+    data = DataFrame({'a': range(DEFAULT_MAX_DEBUG_DATAFRAME_ROWS + 1)})
+
+    message = build_dataframe_debug_message('payload:', data)
+
+    assert 'skipped because dataframe has' in message
+    assert str(data.shape) in message
+    assert data.to_csv() not in message
+
+
+def test_the_row_limit_is_configurable():
+    data = DataFrame({'a': [1, 2, 3]})
+
+    message = build_dataframe_debug_message('payload:', data, max_rows=1)
+
+    assert 'skipped because dataframe has 3 rows' in message
+    assert '(max: 1)' in message

tests/laborious/utils/test_time.py

@@ -0,0 +1,19 @@
+from datetime import UTC, datetime
+
+from laborious.utils.time import to_naive
+
+
+def test_to_naive_returns_naive_input_unchanged() -> None:
+    moment = datetime(2024, 1, 1, 12, 30, 45)
+
+    assert to_naive(moment) == moment
+    assert to_naive(moment).tzinfo is None
+
+
+def test_to_naive_strips_tzinfo_from_aware_input() -> None:
+    moment = datetime(2024, 1, 1, 12, 30, 45, tzinfo=UTC)
+
+    result = to_naive(moment)
+
+    assert result.tzinfo is None
+    assert result == datetime(2024, 1, 1, 12, 30, 45)

tests/laborious/worker/test_worker.py

@@ -48,21 +48,25 @@ def _build_fake_worker(async_result=None, async_error: Exception | None = None):
 
 @patch('laborious.worker.worker.start_http_server')
 def test_start_prometheus_server_success(mock_start_http):
+    logger = MagicMock()
     with patch.object(worker.metrics.APP_UP, 'labels') as labels:
         gauge = MagicMock()
         labels.return_value = gauge
         with patch('laborious.worker.worker.os.getenv', return_value='9090'):
-            worker.start_prometheus_server()
+            worker.start_prometheus_server(logger, {})
     mock_start_http.assert_called_once_with(9090)
     gauge.set.assert_called_once_with(1)
+    logger.custom_info.assert_called_once_with('Prometheus server started on port 9090.', {})
 
 
 @patch('laborious.worker.worker.start_http_server', side_effect=RuntimeError('nope'))
 def test_start_prometheus_server_error_exits(_mock_start_http):
+    logger = MagicMock()
     with patch('laborious.worker.worker.os._exit', side_effect=SystemExit(1)) as m_exit:
         with raises(SystemExit):
-            worker.start_prometheus_server()
+            worker.start_prometheus_server(logger, {})
     m_exit.assert_called_once_with(1)
+    logger.custom_critical.assert_called_once_with('Failed to start Prometheus server: nope', {})
 
 
 @mark.asyncio
@@ -181,6 +185,9 @@ async def test_main_success_exit_zero(monkeypatch):
     prepare_calls = m_prepare.call_args_list
     assert prepare_calls[0].kwargs['runtime'] == 'single'
     assert prepare_calls[3].kwargs['runtime'] == 'single'
+    # Drift and SimpleMetrics load no model code, so every runtime shares the `core` queues.
+    assert prepare_calls[1].kwargs['runtime'] == 'core'
+    assert prepare_calls[2].kwargs['runtime'] == 'core'
 
 
 @mark.asyncio
@@ -241,3 +248,45 @@ async def test_main_worker_gather_error_exits_one(monkeypatch):
             await worker.main()
 
     assert m_logger.return_value.custom_error.called
+
+
+@mark.asyncio
+async def test_main_legacy_runtime_installs_single(monkeypatch):
+    """The `legacy` runtime is served by the `single` store runtime."""
+    monkeypatch.setenv('RUNTIME', 'legacy')
+    fake_plugin = MagicMock()
+    fake_plugin.install_runtime = AsyncMock(side_effect=RuntimeError('stop after install'))
+    with (
+        patch('laborious.worker.worker.start_prometheus_server'),
+        patch('laborious.worker.worker.get_logger'),
+        patch(
+            'laborious.worker.worker.build_mongodb_config',
+            return_value={'connection_string': 'cs', 'database_name': 'db'},
+        ),
+        patch('laborious.worker.worker.NotificationHandler'),
+        patch('laborious.worker.worker.MetricsController'),
+        patch(
+            'laborious.worker.worker.build_plugin_store_config',
+            return_value={
+                'base_url': '',
+                'owner': '',
+                'repo': '',
+                'username': None,
+                'password': None,
+                'branch': None,
+                'cache_ttl_seconds': None,
+                'pypi_index_url': '',
+                'pypi_username': None,
+                'pypi_password': None,
+            },
+        ),
+        patch('laborious.worker.worker.PluginStore', return_value=fake_plugin),
+        patch.object(worker.metrics.APP_UP, 'labels') as labels,
+        patch('laborious.worker.worker.sys.exit', side_effect=SystemExit(1)),
+    ):
+        labels.return_value = MagicMock()
+        with raises(SystemExit):
+            await worker.main()
+
+    fake_plugin.install_runtime.assert_awaited_once()
+    assert fake_plugin.install_runtime.await_args.kwargs['runtime_name'] == 'single'

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -326,6 +326,7 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
                     'model_id': input_data['model_id'],
                     'prediction_confidence': input_data['prediction_confidence'],
                     'comment': input_data['comment'],
+                    'metadata_id': None,
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
@@ -389,6 +390,56 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
     assert workflow_mock.execute_local_activity_method.call_count == 1
 
 
+@mark.asyncio
+@patch(
+    'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
+    new_callable=AsyncMock,
+)
+async def test_run_default_path_flag_passes_metadata_id(
+    workflow_mock, format_and_export_prediction
+):
+    # Arrange: metadata_id reaches format_default_prediction, which is the row that ends up in
+    # Postgres via export_data_to_postgres (CONTINUE/REPEAT path).
+    input_data = {
+        'metadata': metadata,
+        'path_flag': 'default',
+        'data': {'test': 'data'},
+        'timestamp': '2021-01-01',
+        'model_id': 1,
+        'model_name': metadata['metadata']['model_name'],
+        'prediction_confidence': 0,
+        'schema': 'test_schema',
+        'table_name': 'test_table',
+        'opc_output_config': {'test': 'config'},
+        'comment': 'test_comment',
+        'metadata_id': 'a' * 24,
+    }
+
+    workflow_mock.execute_activity_method.side_effect = [
+        (MagicMock(), MagicMock()),
+        MagicMock(),
+        MagicMock(),
+    ]
+
+    # Act
+    await format_and_export_prediction.run(input_data)
+
+    # Assert
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': 'a' * 24,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch(
     'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
@@ -638,6 +689,20 @@ async def test_run_default_path_flag_with_pi_web_api(workflow_mock, format_and_e
 
     await format_and_export_prediction.run(input_data)
 
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
     workflow_mock.execute_activity_method.assert_has_calls(
         [
             call(

tests/laborious/workflows/subworkflows/test_prediction_process.py

@@ -57,7 +57,7 @@ async def test_run(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -217,7 +217,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process):
 
     # Mock the activity responses
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('stop', 0.95, 'Input data with bad quality'),  # input_gate
+        ('stop', 0.95, 'Input data with bad quality', None),  # input_gate
     ]
 
     # Act
@@ -275,7 +275,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('repeat', 0.95, 'Input data with bad quality'),  # input_gate
+        ('repeat', 0.95, 'Input data with bad quality', None),  # input_gate
         ('continue', 0.95, 'Error'),  # mlflow_response_gate (transform)
     ]
 
@@ -365,7 +365,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -476,7 +476,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -598,6 +598,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -616,18 +617,22 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
-    # Assert
+    # Assert: STOP never reaches a write, even though it now gets a `document` same as CONTINUE
+    # and REPEAT do — it just never has anywhere to use it.
     assert result is True
     workflow_mock.execute_local_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_not_called()
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
-    # Arrange
+    # Arrange: REPEAT now writes metadata just like CONTINUE — it does insert a new `predictions`
+    # row (reusing the previous row's values), so it needs a `metadata_id` too.
     data = {'test': 'data'}
     path_flag = 'repeat'
     confidence = 0.95
@@ -637,6 +642,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -655,11 +661,20 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
     # Assert
     assert result is True
-    workflow_mock.execute_activity_method.assert_called_once_with(
+    metadata_id = workflow_mock.execute_activity_method.return_value
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+    workflow_mock.execute_activity_method.assert_any_call(
         Activities.repeat_last_prediction,
         {
             **metadata,
@@ -667,6 +682,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
             'table_name': table_name,
             'model': model,
             'last_timestamp': last_timestamp,
+            'metadata_id': metadata_id,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
@@ -674,6 +690,56 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_repeat_no_document_skips_write_activity_call(
+    workflow_mock, prediction_process
+):
+    # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today,
+    # `mlflow_response_gate`/`mlflow_content_gate`) must not even dispatch the remote
+    # write_prediction_metadata activity — that would be a wasted Temporal round-trip on every
+    # REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01).
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    called_activities = [c.args[0] for c in workflow_mock.execute_activity_method.call_args_list]
+    assert Activities.write_prediction_metadata not in called_activities
+    # repeat_last_prediction still happens, just with metadata_id=None.
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_continue(workflow_mock, prediction_process):
@@ -688,6 +754,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
     prediction_store_policy = 'erl:1'
+    document = {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -713,11 +782,18 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         'Prediction Process',
+        document,
     )
 
     # Assert
+    metadata_id = workflow_mock.execute_activity_method.return_value
     assert result is True
-    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_called_once_with(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
     workflow_mock.execute_child_workflow.assert_called_once_with(
         'subworkflow.format_and_export_prediction',
         {
@@ -733,6 +809,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
             'table_name': table_name,
             'transform_table_name': 'test_transform_table',
             'comment': 'Prediction Process',
+            'metadata_id': metadata_id,
             'opc_output_config': {'test': 'config'},
             'pi_web_api_output_config': {
                 'endpoint': 'https://test-pi-server.com',
@@ -745,6 +822,153 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     )
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_no_document_skips_write_activity_call(
+    workflow_mock, prediction_process
+):
+    # Arrange: same gap as the REPEAT case above (review finding, 2026-10-01) — no document must
+    # not dispatch the remote write_prediction_metadata activity.
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: write_prediction_metadata's retry_policy has maximum_attempts=1 — a timeout, an
+    # unavailable worker, or any dispatch failure raises straight into the workflow with no
+    # retry. Business rule 1 says this must never block the prediction write (review finding,
+    # 2026-10-01) — confirm the CONTINUE branch still proceeds instead of letting it propagate.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = RuntimeError('activity dispatch failed')
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: the prediction write still happens, with metadata_id=None, instead of raising.
+    assert result is True
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_repeat_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: only the write_prediction_metadata call (first) fails — repeat_last_prediction
+    # (second) must still run normally; its own failure is a separate, intentional
+    # non-best-effort path, not what this test is about.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = [
+        RuntimeError('activity dispatch failed'),
+        MagicMock(),
+    ]
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: repeat_last_prediction still runs, with metadata_id=None, instead of raising.
+    assert result is True
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
@@ -791,6 +1015,30 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_run_cleans_up_and_reraises_on_pipeline_failure(workflow_mock, prediction_process):
+    """A failure anywhere in the pipeline must still release the MinIO offload before the
+    workflow fails — cleanup is not conditional on success."""
+    prediction_process._run_prediction_pipeline = AsyncMock(side_effect=RuntimeError('boom'))
+    input_data = {
+        'metadata': metadata,
+        'data': {'some': 'data'},
+        'model_id': 1,
+        'model_name': 'test_model_name',
+    }
+
+    with raises(RuntimeError, match='boom'):
+        await prediction_process.run(input_data)
+
+    workflow_mock.execute_activity_method.assert_called_once_with(
+        Activities.cleanup_minio_objects_expired,
+        {**metadata, 'data': input_data['data']},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
@@ -826,7 +1074,7 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'ok'),
+        ('continue', 0.95, 'ok', None),  # input_gate
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),

tests/laborious/workflows/test_minimal_retrain.py

@@ -1,6 +1,7 @@
 from unittest.mock import ANY, AsyncMock, call, patch
 
-from pytest import fixture, mark
+from pytest import fixture, mark, raises
+from temporalio.exceptions import ApplicationError
 
 from laborious.activities.activities import Activities
 from laborious.workflows.minimal_retrain import MinimalRetrain
@@ -192,11 +193,12 @@ async def test_run_storage_fail(workflow_mock: AsyncMock, minimal_retrain: Minim
         ]
     )
 
-    from pytest import raises
-
-    with raises(ValueError, match='No data returned from query'):
+    with raises(ApplicationError, match='No data returned from query') as exc_info:
         await minimal_retrain.run(input_data)
 
+    assert exc_info.value.non_retryable is True
+    assert exc_info.value.type == 'NoDataReturnedError'
+
     workflow_mock.execute_activity_method.assert_called_once_with(
         Activities.load_query_with_minio_offload,
         {

values.yaml

@@ -1,319 +0,0 @@
-#
-# Default values for sientia-laborious-worker using the sientia-module chart (0.6.x).
-# This is a YAML-formatted file.
-# Declare variables to be passed into your templates.
-#
-
-projectName: &projectName "sientia-laborious-worker"
-
-# -----------------------------------------------------------------------------
-# Global configuration shared by all runtimes
-# -----------------------------------------------------------------------------
-global:
-  namespace: sientia
-
-  image:
-    repository: aignosi.azurecr.io/sientia-module
-    pullPolicy: Always
-    tag: "1.2.0"
-
-  commonLabels: {}
-
-  resources:
-    # Resource limits and requests are important for ResourceBasedTuner to work correctly.
-    # The tuner monitors system CPU and memory usage, so proper resource limits must be set.
-    limits:
-      cpu: 2000m
-      memory: 20Gi
-    requests:
-      cpu: 1000m
-      memory: 2Gi
-
-  livenessProbe:
-    exec:
-      command:
-        - sh
-        - -c
-        - |
-          curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
-    initialDelaySeconds: 1260
-    periodSeconds: 15
-    timeoutSeconds: 5
-    failureThreshold: 3
-
-  readinessProbe:
-    exec:
-      command:
-        - sh
-        - -c
-        - |
-          curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1'
-    initialDelaySeconds: 1200
-    periodSeconds: 10
-    timeoutSeconds: 3
-    failureThreshold: 2
-
-  autoscaling:
-    enabled: false
-    minReplicas: 1
-    maxReplicas: 100
-    targetCPUUtilizationPercentage: 80
-    # targetMemoryUtilizationPercentage: 80
-
-  # Environment variables shared by all runtimes.
-  env:
-    # Entrypoint variables
-    - name: GITHUB_REPO_URL
-      value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
-    - name: GITHUB_BRANCH
-      value: "release/SIENTIAPDE-1646"
-    - name: PYTHON_APP
-      value: "laborious.worker.worker"
-    - name: PYPI_SERVER
-      value: "http://library-distribution-server.library.svc.cluster.local:5000"
-
-    # Application variables
-    - name: POSTGRES_HOST
-      value: "paradedb-rw.paradedb.svc.cluster.local"
-    - name: POSTGRES_PORT
-      value: "5432"
-    - name: POSTGRES_USER
-      value: "postgres"
-    - name: POSTGRES_PASSWORD
-      value: "nFqc81y6kwmr2zuAIx43DhiOosFCVPpeEfTtTWZflkNjB2j1KtEeIANkhFR9mAX3"
-    - name: POSTGRES_DBNAME
-      value: "sientia"
-    - name: POSTGRES_MIN_CONNECTIONS
-      value: "20"
-    # max_connections = number_of_workers * max_concurrent_activities * safety_factor
-    # Example: 4 workers * 50 activities * 0.5 = 100 connections
-    - name: POSTGRES_MAX_CONNECTIONS
-      value: "100"
-
-    - name: MLFLOW_URL
-      value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local:80"
-    - name: MLFLOW_USERNAME
-      value: "aignosi"
-    - name: MLFLOW_PASSWORD
-      value: "1L0FP50j3ncp123"
-
-    # Plugin store (model-library-store Git + runtime packages).
-    - name: STORE_BASE_URL
-      value: "http://gitea-http.gitea.svc.cluster.local:3000"
-    - name: STORE_OWNER
-      value: "aignosi"
-    - name: STORE_REPO
-      value: "suse-model-store"
-    - name: STORE_USERNAME
-      valueFrom:
-        secretKeyRef:
-          name: sientia-plugin-store-credentials
-          key: username
-    - name: STORE_PASSWORD
-      valueFrom:
-        secretKeyRef:
-          name: sientia-plugin-store-credentials
-          key: password
-    - name: STORE_CACHE_TTL_SECONDS
-      value: "3600"
-
-    - name: OPC_ID
-      value: "1"
-    - name: OPC_SERVER_NAME
-      value: "default_server"
-    - name: OPC_URL
-      value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840"
-
-    - name: LOG_LEVEL
-      value: "DEBUG"
-    - name: HTTP_METRICS_PORT
-      value: "9090"
-    - name: HTTP_SDK_METRICS_PORT
-      value: "9091"
-    - name: PROJECT_NAME
-      value: "sientia-laborious"
-
-    - name: TEMPORAL_HOST
-      value: "temporal-frontend.temporal.svc.cluster.local:7233"
-    - name: TEMPORAL_NAMESPACE
-      value: "laborious"
-
-    - name: MONGODB_USERNAME
-      value: "root"
-    - name: MONGODB_PASSWORD
-      value: "wKZDbMNU1c"
-    - name: MONGODB_URL
-      value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
-    - name: MONGODB_DATABASE
-      value: "sientia"
-    - name: MONGODB_TTL_INDEX_HOURS
-      value: "1"
-
-    - name: MINIO_ENDPOINT_URL
-      value: "minio.minio.svc.cluster.local:9000"
-    - name: MINIO_ACCESS_KEY
-      value: "admin"
-    - name: MINIO_SECRET_KEY
-      value: "LiArt4eNmJ"
-    - name: MINIO_DEFAULT_BUCKET
-      value: "sientia"
-    - name: MINIO_RETENTION_HOURS
-      value: "24"
-    - name: SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES
-      value: "0.5"
-
-    # Temporal worker tuning for PredictionsBatch.
-    # IMPORTANT: prefix must be PREDICTIONSBATCH_ (from class name PredictionsBatch).
-    - name: PREDICTIONSBATCH_MAX_CONCURRENT_WORKFLOW_TASKS
-      value: "20"
-    - name: PREDICTIONSBATCH_MAX_CONCURRENT_ACTIVITIES
-      value: "60"
-    - name: PREDICTIONSBATCH_ACTIVITY_EXECUTOR_MAX_WORKERS
-      value: "10"
-    - name: PREDICTIONSBATCH_MAX_CONCURRENT_LOCAL_ACTIVITIES
-      value: "20"
-    - name: PREDICTIONSBATCH_MAX_CACHED_WORKFLOWS
-      value: "200"
-    - name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_MINIMUM
-      value: "3"
-    - name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_INITIAL
-      value: "5"
-    - name: PREDICTIONSBATCH_WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM
-      value: "15"
-    - name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_MINIMUM
-      value: "3"
-    - name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_INITIAL
-      value: "10"
-    - name: PREDICTIONSBATCH_ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM
-      value: "30"
-
-    - name: MINIMALRETRAIN_MAX_CONCURRENT_ACTIVITIES
-      value: "5"
-    - name: MINIMALRETRAIN_ACTIVITY_EXECUTOR_MAX_WORKERS
-      value: "5"
-    - name: MINIMALRETRAIN_MAX_CONCURRENT_LOCAL_ACTIVITIES
-      value: "5"
-    - name: MINIMALRETRAIN_MAX_CACHED_WORKFLOWS
-      value: "5"
-    - name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_MINIMUM
-      value: "5"
-    - name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_INITIAL
-      value: "5"
-    - name: MINIMALRETRAIN_WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM
-      value: "5"
-    - name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_MINIMUM
-      value: "5"
-    - name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_INITIAL
-      value: "5"
-    - name: MINIMALRETRAIN_ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM
-      value: "5"
-
-    - name: PI_WEB_API_BASE_URL
-      value: "https://pivision.votorantimcimentos.com/piwebapi"
-    - name: PI_WEB_API_AUTH_TYPE
-      value: "basic"
-    - name: PI_WEB_API_AUTH_TOKEN
-      valueFrom:
-        secretKeyRef:
-          name: pi-web-api-auth-token
-          key: token
-
-    # Thread-pool size for non-runtime workers that also use prepare_worker.
-    - name: SIMPLEMETRICS_ACTIVITY_EXECUTOR_MAX_WORKERS
-      value: "20"
-    - name: DRIFT_ACTIVITY_EXECUTOR_MAX_WORKERS
-      value: "20"
-
-# -----------------------------------------------------------------------------
-# Runtimes configuration
-# -----------------------------------------------------------------------------
-# IMPORTANT:
-# - The runtime name is used by the worker bootstrap to resolve plugins and task queues.
-# - Keep runtime names in sync with the plugin-store runtime names.
-runtimes:
-  basic:
-    replicas: 1
-  legacy:
-    replicas: 1
-    env:
-      - name: "GITHUB_BRANCH"
-        value: "main"
-      - name: "MLFLOW_HOST"
-        value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local"
-      - name: "MLFLOW_PORT"
-        value: "80"
-
-# -----------------------------------------------------------------------------
-# Chart-level configuration (applies to all runtimes)
-# -----------------------------------------------------------------------------
-imagePullSecrets:
-  - name: docker-hub-secret
-
-nameOverride: *projectName
-fullnameOverride: *projectName
-
-serviceAccount:
-  create: true
-  automount: true
-  annotations: {}
-  name: *projectName
-
-podAnnotations: {}
-podLabels: {}
-
-podSecurityContext: {}
-securityContext: {}
-
-volumes: []
-volumeMounts: []
-
-nodeSelector: {}
-tolerations: []
-affinity: {}
-
-services:
-  sdk-metrics:
-    enabled: true
-    type: ClusterIP
-    port: 9091
-    targetPort: 9091
-    name: sdk-metrics
-  metrics:
-    enabled: true
-    type: ClusterIP
-    port: 9090
-    targetPort: 9090
-    name: metrics
-
-# Configuração do ServiceMonitor para o Prometheus Operator
-# ref: https://github.com/prometheus-operator/prometheus-operator
-serviceMonitor:
-  enabled: true
-  endpoints:
-    - port: metrics
-      path: /metrics
-      interval: 30s
-      relabelings: []
-    - port: sdk-metrics
-      path: /metrics
-      interval: 30s
-      relabelings: []
-  additionalLabels:
-    release: kube-prometheus-stack
-
-ssh:
-  enabled: true
-  secretName: git-ssh-key-sientia-laborious-worker
-  sshPath: /mnt/.ssh
-  knownHostsPath: /mnt/known_hosts
-
-# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=<pwd>
-#
-# helm upgrade --install sientia-laborious-worker /home/grezewave/Documents/projects/sientia/sientia-core-applications/sientia-module -n sientia --create-namespace -f ./values.yaml
-
-# kubectl create secret generic git-ssh-key-sientia-laborious-worker \
-#   --namespace sientia \
-#   --from-file=ssh-privatekey=git_key \
-#   --type=kubernetes.io/ssh-auth
-
-# helm upgrade --install sientia-laborious-worker /home/grezewave/Documents/projects/sientia/sientia-core-applications/sientia-module -n sientia --create-namespace -f ./values.yaml