Verify if repository is production-ready

[mikkihugo]

User description

What does this PR do?

Brief description of the changes made in this pull request.

Related Issues

Fixes #123
or
Related to #456

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Testing

• Tests added for new functionality
• All tests passing locally (mix test)
• Code quality checks passing (mix quality)

Checklist

• Code follows the style guidelines (run mix format)
• Documentation updated (if user-facing change)
• CHANGELOG.md updated (for new features/breaking changes)
• No new warnings introduced (mix credo --strict)
• Types are correct (Dialyzer happy)
• No breaking changes (or documented in CHANGELOG)

Screenshot (if applicable)

N/A

---

PR Type

Enhancement, Documentation

---

Description

• Workflow Lifecycle Management: Added five new functions (cancel_workflow_run/3, list_workflow_runs/2, retry_failed_workflow/3, pause_workflow_run/2, resume_workflow_run/2) for production-grade workflow control and monitoring

• Lineage Tracking System: Introduced new Lineage module for DAG-based execution history tracking, deterministic workflow replay, and evolutionary learning capabilities

• Multi-Tenancy Support: Added tenant_id fields to core schemas (WorkflowRun, StepState, StepTask) with database migration including indexes and RLS policy templates

• Distributed Backend Refactoring: Converted DistributedBackend from stub to functional implementation delegating to ObanBackend with PostgreSQL pgmq-based execution

• Execution Strategy Simplification: Renamed execution modes from :sync/:oban/:distributed to :local/:distributed for clearer API semantics

• Comprehensive Documentation: Added API reference guide, updated README and getting started guide, created documentation index, and clarified library positioning

• Version Update: Bumped version to 0.1.5 with corresponding changelog entries

• Development Environment: Updated database naming from quantum_flow to singularity_workflow throughout configuration files

• Cleanup: Removed obsolete migrations, tests, and documentation files related to schema renaming and old architecture

---

Diagram Walkthrough

flowchart LR
  A["Core Schemas<br/>WorkflowRun, StepState, StepTask"] -->|"add tenant_id"| B["Multi-Tenancy<br/>Support"]
  C["Executor Module"] -->|"new functions"| D["Workflow Lifecycle<br/>Management"]
  E["New Lineage Module"] -->|"DAG tracking"| F["Evolutionary<br/>Learning"]
  G["DistributedBackend"] -->|"refactored to"| H["PostgreSQL pgmq<br/>Execution"]
  I["Execution Strategy"] -->|"rename modes"| J["Cleaner API<br/>local/distributed"]
  D --> K["Production-Ready<br/>Operations"]
  B --> K
  F --> K
  H --> K
  J --> K

Loading

File Walkthrough

	Relevant files
Enhancement	9 files executor.ex Workflow lifecycle management functions for production operations lib/singularity_workflow/executor.ex Added cancel_workflow_run/3 to cancel running workflows with custom reasons and force options Added list_workflow_runs/2 to query workflows with filtering, pagination, and ordering Added retry_failed_workflow/3 to retry failed workflows with optional step skipping Added pause_workflow_run/2 and resume_workflow_run/2 for soft pause/resume functionality Added private cancel_oban_jobs_for_run/2 helper for distributed job cancellation +402/-0  lineage.ex Lineage tracking module for evolutionary learning systems lib/singularity_workflow/lineage.ex New module for DAG-based lineage tracking and evolutionary memory Implements get_lineage/2 to retrieve complete execution history with task graphs and metrics Implements replay/3 for deterministic workflow reproduction Implements get_lineages/2 and query_lineages/2 for batch lineage queries with filtering Provides helper functions for task graph reconstruction and execution trace building +325/-0  distributed_backend.ex Distributed backend implementation using PostgreSQL pgmq  lib/singularity_workflow/execution/backends/distributed_backend.ex Refactored from stub implementation to functional distributed backend Now delegates to ObanBackend internally for PostgreSQL + pgmq-based execution Updated documentation to clarify Oban is an implementation detail, not exposed to users Implements execute/4 for distributed task execution with resource allocation Implements available?/0 to check if Oban is loaded +59/-95  singularity_workflow.ex Public API delegation and documentation updates                    lib/singularity_workflow.ex Added documentation section for workflow lifecycle management with code examples Updated real-time messaging documentation to clarify PostgreSQL NOTIFY replaces NATS Added defdelegate calls for lifecycle functions: cancel_workflow_run, list_workflow_runs, retry_failed_workflow, pause_workflow_run, resume_workflow_run, get_run_status Updated version from "0.1.0" to "0.1.5" +60/-12  strategy.ex Execution strategy refactoring with cleaner API                    lib/singularity_workflow/execution/strategy.ex Renamed execution modes from :sync/:oban/:distributed to :local/:distributed Updated documentation to clarify Oban is internal implementation detail Simplified execute/4 to only support :local and :distributed modes Updated available?/1 to check DistributedBackend.available?() instead of hardcoded false +21/-18  workflow_definition.ex Execution mode naming consistency updates                                lib/singularity_workflow/dag/workflow_definition.ex Updated all execution mode defaults from :sync to :local throughout the module Updated type specs to reflect :local | :distributed instead of :sync | :oban | :distributed Changed default execution mode in step definitions and workflow defaults +5/-5      workflow_run.ex Multi-tenancy field addition to workflow runs                        lib/singularity_workflow/workflow_run.ex Added tenant_id field (binary_id, nullable) to WorkflowRun schema Updated type spec to include tenant_id: Ecto.UUID.t() | nil Added tenant_id to changeset cast list for multi-tenancy support +4/-0      step_task.ex Multi-tenancy field addition to step tasks                              lib/singularity_workflow/step_task.ex Added tenant_id field (binary_id) to StepTask schema Updated type spec to include tenant_id: Ecto.UUID.t() | nil Added tenant_id to changeset cast list for multi-tenancy support +3/-0      step_state.ex Multi-tenancy field addition to step states                            lib/singularity_workflow/step_state.ex Added tenant_id field (binary_id) to StepState schema Updated type spec to include tenant_id: Ecto.UUID.t() | nil Added tenant_id to changeset cast list for multi-tenancy support +3/-0
Configuration changes	3 files 20251109000000_add_tenant_id_to_all_tables.exs Multi-tenancy support via tenant_id fields and indexes      priv/repo/migrations/20251109000000_add_tenant_id_to_all_tables.exs New migration adding tenant_id UUID field to workflow_runs, workflow_step_states, and workflow_step_tasks Creates indexes on tenant_id for query performance and composite indexes for common queries Includes optional Row-Level Security (RLS) policy templates for multi-tenancy enforcement Adds database comments explaining tenant isolation strategy +134/-0  mix.exs Version number correction                                                                mix.exs Updated version from "1.0.2" to "0.1.5" to reflect actual release version +1/-1      flake.nix Development environment database naming updates                    flake.nix Updated database name from quantum_flow to singularity_workflow throughout Updated environment variable references and echo messages to use correct database name Updated development environment setup messages +6/-6
Documentation	8 files notifications.ex Documentation updates for messaging infrastructure clarity lib/singularity_workflow/notifications.ex Updated module documentation to clarify PostgreSQL NOTIFY is a NATS replacement Changed terminology from "notifications" to "messages" throughout documentation Clarified that pgmq provides message persistence and reliability Updated benefits section to emphasize NATS replacement capability +11/-9    Evo.txt Evolutionary learning package specification document          Evo.txt New comprehensive specification for singularity_evolution package (separate from core library) Defines adaptive planner for LLM-based goal decomposition with learned patterns Specifies evolution engine for fitness-based planner improvement and hot-reload Includes architecture, module specifications, integration points with lineage API Provides implementation roadmap, testing strategy, and deployment guidance +713/-0  API_REFERENCE.md Complete API reference documentation with examples              docs/API_REFERENCE.md New comprehensive API reference documentation covering all public APIs Documents workflow execution, lifecycle management, messaging, HTDAG orchestration Includes Phoenix LiveView and Channels integration examples Provides comparison tables and decision guidance for API selection Covers execution strategies (local vs distributed) with use case guidance +719/-0  README.md README updates for library positioning and features            README.md Clarified this is a library package, not a standalone application Updated installation instructions to reference version "0.1.5" Added workflow lifecycle management section with examples Added Phoenix integration section with LiveView/Channels examples Updated messaging terminology from "notifications" to "messaging" Clarified deployment section is for applications using the library Added comparison table for Singularity.Workflow vs Phoenix.PubSub +149/-59 GETTING_STARTED.md Getting started guide updates for library usage                    GETTING_STARTED.md Clarified this is a library package to be added as a dependency Updated installation version to "0.1.5" Simplified database setup to use application's existing repo Removed separate Singularity.Workflow.Repo configuration Updated first workflow example to use modern __workflow_steps__ pattern Clarified migrations are optional and managed by library +40/-47  README.md Documentation index and navigation guide                                  docs/README.md New documentation index and navigation guide Organizes all documentation by category (getting started, core, features, community) Provides quick navigation based on user goals Explains documentation philosophy and principles Includes links to external resources and versioning information +123/-0  SECURITY.md Security documentation cleanup                                                      SECURITY.md Removed reference to non-existent SECURITY_AUDIT.md file Updated security audit statement to be more general +1/-1      CHANGELOG.md Release notes for workflow lifecycle management features  CHANGELOG.md Added new version [0.1.5] release notes with workflow lifecycle management features Documented five new workflow control functions: cancel_workflow_run/3, list_workflow_runs/2, retry_failed_workflow/3, pause_workflow_run/2, and resume_workflow_run/2 Enhanced documentation section with new API reference and updated existing guides Reorganized and clarified documentation entries, including renaming SINGULARITY_WORKFLOW_REFERENCE.md to API_REFERENCE.md +19/-5
Tests	1 files singularity_workflow_doctest.exs Doctest infrastructure for documentation examples                test/singularity_workflow_doctest.exs New test file for running doctests from main module and executor Includes commented-out doctests for database-dependent modules Provides template for testing documentation examples +13/-0
Miscellaneous	1 files .envrc Development environment comment update                                      .envrc Updated comment from quantum_flow to singularity-workflows for clarity +1/-1
Additional files	12 files CODEOWNERS +1/-1      GITHUB_REPOSITORY_SETUP.md +0/-325  QUANTUM_FLOW_REFERENCE.md +0/-359  RELEASE_PROCESS.md +0/-193  SCHEMA_MIGRATION_GUIDE.md +0/-141  SECURITY_AUDIT.md +0/-247  architecture_diagrams.md +0/-447  README.md +0/-171  20251103234710_rename_quantumflow_schema_to_singularity_workflow.exs +0/-60    SNAPSHOT_TESTING.md +0/-157  executor_test.exs.old +0/-553  schema_rename_migration_test.exs +0/-410

---

[qodo-code-review]

PR Compliance Guide 🔍

(Compliance updated until commit singularity-ng/singularity-workflows@04c318f)

Below is a summary of compliance checks for this PR:

Security Compliance
⚪	Insecure job cancellation Description: Cancelling Oban jobs queries and cancels by matching args->>'workflow_run_id' without namespacing or tenant checks, which may cancel unrelated jobs if args are uncontrolled or schema mismatch occurs; verify args structure and restrict scope. executor.ex [739-778] Referred Code defp cancel_oban_jobs_for_run(run_id, repo) do import Ecto.Query try do # Query Oban jobs table for this workflow run oban_config = Application.get_env(:singularity, Oban, []) oban_repo = Keyword.get(oban_config, :repo, repo) if function_exported?(oban_repo, :all, 1) do query = from(j in "oban_jobs", where: fragment("?->>'workflow_run_id' = ?", j.args, ^run_id), where: j.state in ["available", "scheduled", "executing", "retryable"], select: j.id ) job_ids = oban_repo.all(query) # Cancel each job using Oban API Enum.each(job_ids, fn job_id -> case Oban.cancel_job(job_id) do ... (clipped 19 lines)
	State confusion risk Description: Pause/resume uses status "paused" in tasks and sets run.error_message to "PAUSED" without schema-enforced states, enabling spoofing/logic confusion if other parts rely on error_message for real errors; consider explicit status field or guardrails. executor.ex [610-701] Referred Code @spec pause_workflow_run(Ecto.UUID.t(), module()) :: :ok | {:error, term()} def pause_workflow_run(run_id, repo) do import Ecto.Query repo.transaction(fn -> case repo.get(Singularity.Workflow.WorkflowRun, run_id) do nil -> repo.rollback({:error, :not_found}) run -> if run.status != "started" do repo.rollback({:error, {:not_running, run.status}}) end # Update workflow status to paused (custom status) # Note: Schema only has started/completed/failed, so we store in error_message run |> Ecto.Changeset.change(%{ error_message: "PAUSED", updated_at: DateTime.utc_now() }) ... (clipped 71 lines)
	Missing RLS enforcement Description: RLS policies are provided but commented out, and tenant_id is nullable, which can lead to cross-tenant data exposure unless application-level scoping is rigorously enforced. 20251109000000_add_tenant_id_to_all_tables.exs [77-106] Referred Code # Enable Row-Level Security (RLS) - OPTIONAL, commented out for gradual adoption # Uncomment these lines to enforce tenant isolation at database level: # execute "ALTER TABLE workflow_runs ENABLE ROW LEVEL SECURITY" # execute "ALTER TABLE workflow_step_states ENABLE ROW LEVEL SECURITY" # execute "ALTER TABLE workflow_step_tasks ENABLE ROW LEVEL SECURITY" # execute """ # CREATE POLICY tenant_isolation_workflow_runs ON workflow_runs # USING ( # tenant_id IS NULL OR # tenant_id = current_setting('app.current_tenant_id', true)::uuid # ) # """ # execute """ # CREATE POLICY tenant_isolation_step_states ON workflow_step_states # USING ( # tenant_id IS NULL OR # tenant_id = current_setting('app.current_tenant_id', true)::uuid # ) ... (clipped 9 lines)
	Sensitive information exposure Description: Lineage exposes and returns full task input/output and metrics, potentially leaking sensitive data unless access control is enforced at call sites. lineage.ex [262-279] Referred Code defp build_trace(tasks) do # Build execution trace from tasks Enum.map(tasks, fn task -> %{ task_id: task.id, step_slug: task.step_slug, task_index: task.task_index, input: task.input, output: task.output, status: task.status, attempts: task.attempts_count, max_attempts: task.max_attempts, duration_ms: calculate_task_duration(task), idempotency_key: task.idempotency_key, started_at: task.inserted_at, completed_at: task.updated_at } end)
	Availability misconfiguration Description: Unsupported execution modes now error, and distributed availability relies on Oban presence; if misconfigured, tasks could fail open—ensure validation and fallback paths to prevent denial of service. strategy.ex [41-56] Referred Code @spec execute(function(), any(), execution_config(), map()) :: {:ok, any()} | {:error, term()} def execute(step_fn, input, config, context \\ %{}) do case config.execution do :local -> DirectBackend.execute(step_fn, input, config, context) :distributed -> DistributedBackend.execute(step_fn, input, config, context) other -> {:error, {:unsupported_execution_mode, other}} end end @doc """ Check if an execution mode is available. """ @spec available?(:local | :distributed) :: boolean() def available?(:local), do: true def available?(:distributed), do: DistributedBackend.available?() end
Ticket Compliance
⚪	🎫 No ticket provided Create ticket/issue
Codebase Duplication Compliance
⚪	Codebase context is not defined Follow the guide to enable codebase context checks.
Custom Compliance
🟢	Generic: Meaningful Naming and Self-Documenting Code Objective: Ensure all identifiers clearly express their purpose and intent, making code self-documenting Status: Passed Learn more about managing compliance generic rules or creating your own custom rules
⚪	Generic: Comprehensive Audit Trails Objective: To create a detailed and reliable record of critical system actions for security analysis and compliance. Status: Insufficient auditing: New lifecycle actions (cancel, pause, resume, retry, list) log minimal messages without consistent, structured audit details (actor/user, outcome, timestamps), making it unclear if comprehensive audit trails are recorded. Referred Code Logger.info("Workflow cancelled", run_id: run_id, reason: reason ) :ok end end) |> case do {:ok, result} -> result {:error, reason} -> {:error, reason} end end @doc """ List workflow runs with optional filtering. ## Parameters - `repo` - Ecto repository - `filters` - Filter options (optional) ... (clipped 265 lines) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Robust Error Handling and Edge Case Management Objective: Ensure comprehensive error handling that provides meaningful context and graceful degradation Status: Generic errors: Several functions wrap failures into generic tuples (e.g., {:error, reason}) and repurpose fields (e.g., storing "PAUSED" in error_message) which may obscure edge cases and complicate robust handling. Referred Code @spec retry_failed_workflow(Ecto.UUID.t(), module(), keyword()) :: {:ok, Ecto.UUID.t()} | {:error, term()} def retry_failed_workflow(run_id, repo, opts \\ []) do reset_all = Keyword.get(opts, :reset_all, false) case repo.get(Singularity.Workflow.WorkflowRun, run_id) do nil -> {:error, :not_found} run -> if run.status != "failed" and not reset_all do {:error, {:not_failed, run.status}} else # Get workflow module workflow_module = try do String.to_existing_atom("Elixir.#{run.workflow_slug}") rescue ArgumentError -> nil end ... (clipped 144 lines) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Secure Error Handling Objective: To prevent the leakage of sensitive system information through error messages while providing sufficient detail for internal debugging. Status: Sensitive details risk: Logger calls include dynamic reasons (e.g., inspect(reason), Exception.message(e)) which may expose internal error details in logs if surfaced to end users elsewhere; review log routing and visibility. Referred Code :ok -> Logger.debug("Cancelled Oban job", job_id: job_id, run_id: run_id) {:error, reason} -> Logger.warning("Failed to cancel Oban job", job_id: job_id, run_id: run_id, reason: inspect(reason) ) end end) end rescue e -> Logger.warning("Error cancelling Oban jobs", run_id: run_id, error: Exception.message(e) ) end Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Secure Logging Practices Objective: To ensure logs are useful for debugging and auditing without exposing sensitive information like PII, PHI, or cardholder data. Status: Logs may leak data: Lineage surfaces full task input/output and error_message; if logged or exposed, this could include sensitive data—ensure scrubbing or redaction in any logging or external exposure. Referred Code end defp build_trace(tasks) do # Build execution trace from tasks Enum.map(tasks, fn task -> %{ task_id: task.id, step_slug: task.step_slug, task_index: task.task_index, input: task.input, output: task.output, status: task.status, attempts: task.attempts_count, max_attempts: task.max_attempts, duration_ms: calculate_task_duration(task), idempotency_key: task.idempotency_key, started_at: task.inserted_at, completed_at: task.updated_at } end) end ... (clipped 29 lines) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Security-First Input Validation and Data Handling Objective: Ensure all data inputs are validated, sanitized, and handled securely to prevent vulnerabilities Status: Input validation gaps: Functions accept filters/options and interact with DB without visible validation of values (status, limits, offsets) and Lineage exposes raw inputs/outputs; need confirmation of upstream validation and authorization, especially with new multi-tenant fields. Referred Code @spec list_workflow_runs(module(), keyword()) :: {:ok, [Singularity.Workflow.WorkflowRun.t()]} | {:error, term()} def list_workflow_runs(repo, filters \\ []) do import Ecto.Query query = from(r in Singularity.Workflow.WorkflowRun, select: r ) # Apply filters query = if status = filters[:status] do from(r in query, where: r.status == ^status) else query end query = if workflow_slug = filters[:workflow_slug] do from(r in query, where: r.workflow_slug == ^workflow_slug) else ... (clipped 17 lines) Learn more about managing compliance generic rules or creating your own custom rules

Compliance status legend

🟢 - Fully Compliant
🟡 - Partial Compliant
🔴 - Not Compliant
⚪ - Requires Further Human Verification
🏷️ - Compliance label

---

Previous compliance checks

Compliance check up to commit 04c318f

Security Compliance
⚪	Sensitive information exposure Description: Logs may include sensitive context (e.g., reasons, run_id) and uses Logger.warning/Logger.info without sanitization when canceling Oban jobs, potentially exposing workflow identifiers and error details to logs. executor.ex [744-778] Referred Code oban_config = Application.get_env(:singularity, Oban, []) oban_repo = Keyword.get(oban_config, :repo, repo) if function_exported?(oban_repo, :all, 1) do query = from(j in "oban_jobs", where: fragment("?->>'workflow_run_id' = ?", j.args, ^run_id), where: j.state in ["available", "scheduled", "executing", "retryable"], select: j.id ) job_ids = oban_repo.all(query) # Cancel each job using Oban API Enum.each(job_ids, fn job_id -> case Oban.cancel_job(job_id) do :ok -> Logger.debug("Cancelled Oban job", job_id: job_id, run_id: run_id) {:error, reason} -> Logger.warning("Failed to cancel Oban job", ... (clipped 14 lines)
	Sensitive information exposure Description: Lineage execution trace returns full task input/output which may contain sensitive data; exposing or logging this map externally can leak secrets unless access controls are enforced. lineage.ex [260-278] Referred Code end defp build_trace(tasks) do # Build execution trace from tasks Enum.map(tasks, fn task -> %{ task_id: task.id, step_slug: task.step_slug, task_index: task.task_index, input: task.input, output: task.output, status: task.status, attempts: task.attempts_count, max_attempts: task.max_attempts, duration_ms: calculate_task_duration(task), idempotency_key: task.idempotency_key, started_at: task.inserted_at, completed_at: task.updated_at }
	Multi-tenant data leakage Description: RLS is commented out, so multi-tenant isolation relies on application logic; without enforcing RLS, a mis-scoped query could expose cross-tenant data. 20251109000000_add_tenant_id_to_all_tables.exs [77-106] Referred Code # Enable Row-Level Security (RLS) - OPTIONAL, commented out for gradual adoption # Uncomment these lines to enforce tenant isolation at database level: # execute "ALTER TABLE workflow_runs ENABLE ROW LEVEL SECURITY" # execute "ALTER TABLE workflow_step_states ENABLE ROW LEVEL SECURITY" # execute "ALTER TABLE workflow_step_tasks ENABLE ROW LEVEL SECURITY" # execute """ # CREATE POLICY tenant_isolation_workflow_runs ON workflow_runs # USING ( # tenant_id IS NULL OR # tenant_id = current_setting('app.current_tenant_id', true)::uuid # ) # """ # execute """ # CREATE POLICY tenant_isolation_step_states ON workflow_step_states # USING ( # tenant_id IS NULL OR # tenant_id = current_setting('app.current_tenant_id', true)::uuid # ) ... (clipped 9 lines)
Ticket Compliance
⚪	🎫 No ticket provided Create ticket/issue
Codebase Duplication Compliance
⚪	Codebase context is not defined Follow the guide to enable codebase context checks.
Custom Compliance
🟢	Generic: Meaningful Naming and Self-Documenting Code Objective: Ensure all identifiers clearly express their purpose and intent, making code self-documenting Status: Passed Learn more about managing compliance generic rules or creating your own custom rules
⚪	Generic: Comprehensive Audit Trails Objective: To create a detailed and reliable record of critical system actions for security analysis and compliance. Status: Incomplete auditing: New lifecycle functions log some actions but do not consistently include user identity and full context in all critical operations (cancel/pause/resume/retry/list), which may be insufficient for reconstructing events. Referred Code Logger.info("Workflow cancelled", run_id: run_id, reason: reason ) :ok end end) |> case do {:ok, result} -> result {:error, reason} -> {:error, reason} end end @doc """ List workflow runs with optional filtering. ## Parameters - `repo` - Ecto repository - `filters` - Filter options (optional) ... (clipped 263 lines) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Robust Error Handling and Edge Case Management Objective: Ensure comprehensive error handling that provides meaningful context and graceful degradation Status: Error context gaps: Functions like cancel_workflow_run/pause/resume use transactions but may return generic {:error, reason} without structured context and lack input validation for filters; lineage.replay relies on external modules without clear error propagation. Referred Code @spec list_workflow_runs(module(), keyword()) :: {:ok, [Singularity.Workflow.WorkflowRun.t()]} | {:error, term()} def list_workflow_runs(repo, filters \\ []) do import Ecto.Query query = from(r in Singularity.Workflow.WorkflowRun, select: r ) # Apply filters query = if status = filters[:status] do from(r in query, where: r.status == ^status) else query end query = if workflow_slug = filters[:workflow_slug] do from(r in query, where: r.workflow_slug == ^workflow_slug) else ... (clipped 214 lines) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Secure Error Handling Objective: To prevent the leakage of sensitive system information through error messages while providing sufficient detail for internal debugging. Status: Potential info leak: Logger.warning in cancel_oban_jobs_for_run logs inspected reasons which could include internal details; user-facing exposure is unclear from diff and may require confirmation of log sink and log levels. Referred Code Logger.warning("Failed to cancel Oban job", job_id: job_id, run_id: run_id, reason: inspect(reason) ) end end) end rescue e -> Logger.warning("Error cancelling Oban jobs", run_id: run_id, error: Exception.message(e) ) end end Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Secure Logging Practices Objective: To ensure logs are useful for debugging and auditing without exposing sensitive information like PII, PHI, or cardholder data. Status: Sensitive data risk: Lineage builds traces including full task input/output which may contain sensitive data; no redaction or filtering is shown before exposure through APIs or logs. Referred Code defp build_trace(tasks) do # Build execution trace from tasks Enum.map(tasks, fn task -> %{ task_id: task.id, step_slug: task.step_slug, task_index: task.task_index, input: task.input, output: task.output, status: task.status, attempts: task.attempts_count, max_attempts: task.max_attempts, duration_ms: calculate_task_duration(task), idempotency_key: task.idempotency_key, started_at: task.inserted_at, completed_at: task.updated_at } end) Learn more about managing compliance generic rules or creating your own custom rules
	Generic: Security-First Input Validation and Data Handling Objective: Ensure all data inputs are validated, sanitized, and handled securely to prevent vulnerabilities Status: Missing validation: Query builders accept filter values (status, workflow_slug, order_by, limit/offset) without explicit validation and lineage APIs expose raw input/output fields, which could lead to unsafe data handling if consumed externally. Referred Code # Apply filters query = if status = filters[:status] do from(r in query, where: r.status == ^status) else query end query = if workflow_slug = filters[:workflow_slug] do from(r in query, where: r.workflow_slug == ^workflow_slug) else query end # Apply ordering order_by = filters[:order_by] || {:desc, :inserted_at} query = from(r in query, order_by: ^[order_by]) # Apply pagination limit = filters[:limit] || 100 ... (clipped 8 lines) Learn more about managing compliance generic rules or creating your own custom rules

[mikkihugo]

@copilot revi3w

[Copilot]

@mikkihugo I've opened a new pull request, #8, to work on those changes. Once the pull request is ready, I'll request review from you.

[qodo-code-review]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix critical N+1 query performance issue Fix a critical N+1 query performance issue in get_lineages/2 by fetching all runs, steps, tasks, and dependencies in a single batch query for all run_ids, instead of querying for each run_id individually in a loop. lib/singularity_workflow/lineage.ex [158-169] def get_lineages(run_ids, repo) do + runs = from(r in Singularity.Workflow.WorkflowRun, where: r.id in ^run_ids) |> repo.all() |> Map.new(&{&1.id, &1}) + steps = from(s in Singularity.Workflow.StepState, where: s.run_id in ^run_ids) |> repo.all() |> Enum.group_by(& &1.run_id) + tasks = from(t in Singularity.Workflow.StepTask, where: t.run_id in ^run_ids) |> repo.all() |> Enum.group_by(& &1.run_id) + dependencies = from(d in Singularity.Workflow.StepDependency, where: d.run_id in ^run_ids) |> repo.all() |> Enum.group_by(& &1.run_id) + lineages = Enum.map(run_ids, fn run_id -> - case get_lineage(run_id, repo) do - {:ok, lineage} -> lineage - {:error, _} -> nil - end + run = runs[run_id] + run_steps = Map.get(steps, run_id, []) + run_tasks = Map.get(tasks, run_id, []) + run_deps = Map.get(dependencies, run_id, []) + + # This reuses the existing builder functions but with pre-fetched data + task_graph = build_task_graph(run_steps, run_deps) + execution_trace = build_trace(run_tasks) + metrics = calculate_metrics(run, run_steps, run_tasks) + + %{ + run_id: run_id, + goal: extract_goal(run.input), + workflow_slug: run.workflow_slug, + task_graph: task_graph, + execution_trace: execution_trace, + metrics: metrics, + started_at: run.started_at, + completed_at: run.completed_at, + status: run.status + } end) - |> Enum.reject(&is_nil/1) {:ok, lineages} end Apply / Chat Suggestion importance[1-10]: 9 __ Why: This suggestion correctly identifies a critical N+1 query performance issue that would make the get_lineages function unusable for even a moderate number of run_ids and provides a correct, efficient batch-loading solution.	High
	Use a dedicated field for paused state Add a dedicated paused_at timestamp field to the schema instead of misusing the error_message field to represent the "PAUSED" state, which avoids potential bugs and improves data modeling. lib/singularity_workflow/executor.ex [627-631] -# Update workflow status to paused (custom status) -# Note: Schema only has started/completed/failed, so we store in error_message +# Update workflow status to paused run |> Ecto.Changeset.change(%{ - error_message: "PAUSED", + paused_at: DateTime.utc_now(), updated_at: DateTime.utc_now() }) |> repo.update!() [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 7 __ Why: The suggestion correctly points out that misusing the error_message field for state management is poor design and can lead to bugs, proposing a much cleaner solution with a dedicated paused_at field.	Medium
Security	Prevent atom exhaustion security vulnerability Replace String.to_existing_atom with the safer Module.concat to prevent a potential atom exhaustion security vulnerability when constructing module names from dynamic data. lib/singularity_workflow/executor.ex [557-561] # Get workflow module workflow_module = try do - String.to_existing_atom("Elixir.#{run.workflow_slug}") + Module.concat(["Elixir", run.workflow_slug]) rescue ArgumentError -> nil end [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a potential security risk with String.to_existing_atom and proposes the idiomatic and safer Module.concat as a replacement, which is a critical security hardening practice.	Medium
More

[qodo-code-review]

CI Feedback 🧐

A test triggered by this PR failed. Here is an AI-generated analysis of the failure:

Action: Test & Quality Checks

Failed stage: Run migrations [❌]

Failure summary: The action failed due to database authentication errors when running mix compile --warnings-as-errors and subsequent Ecto tasks: - PostgreSQL connection attempts failed with FATAL 28P01 (invalid_password) password authentication failed for user "runner" (lines 563–567). - Container logs show Role "runner" does not exist and auth method scram-sha-256 (lines 608–621), indicating the DB user configured in the app (runner) was not created in the service DB and/or has no valid password. - This led to DBConnection.ConnectionError: connection not available and request was dropped from queue after 5998ms, during Ecto sandbox checkout and migrations: - db_connection/lib/db_connection/ownership.ex:108 - ecto_sql/lib/ecto/adapters/sql/sandbox.ex:554, :621 - ecto_sql/lib/ecto/migrator.ex:170 - mix/tasks/ecto.migrate.ex:151, :139 Fix by aligning database credentials: create the runner role with a password in the Postgres service, or update the app’s DB config (username/password) to match available roles (e.g., postgres) and ensure PG* env vars are set in the workflow.

Relevant error logs: 1: ##[group]Runner Image Provisioner 2: Hosted Compute Agent ... 447: ==> credo 448: Compiling 251 files (.ex) 449: Generated credo app 450: ==> postgrex 451: Compiling 69 files (.ex) 452: Generated postgrex app 453: ==> ecto_sql 454: Compiling 25 files (.ex) 455: Generated ecto_sql app 456: ==> pgmq 457: Compiling 2 files (.ex) 458: Generated pgmq app 459: ==> oban 460: Compiling 63 files (.ex) 461: Generated oban app 462: ##[group]Run mix compile --warnings-as-errors 463: �[36;1mmix compile --warnings-as-errors�[0m 464: shell: /usr/bin/bash -e {0} ... 548: ==> pgmq 549: Compiling 2 files (.ex) 550: Generated pgmq app 551: ==> oban 552: Compiling 63 files (.ex) 553: Generated oban app 554: ==> nimble_ownership 555: Compiling 2 files (.ex) 556: Generated nimble_ownership app 557: ==> mox 558: Compiling 2 files (.ex) 559: Generated mox app 560: ==> singularity_workflow 561: Compiling 40 files (.ex) 562: Generated singularity_workflow app 563: 16:35:04.204 [error] Postgrex.Protocol (#PID<0.4788.0>) failed to connect: ** (Postgrex.Error) FATAL 28P01 (invalid_password) password authentication failed for user "runner" 564: 16:35:04.204 [error] Postgrex.Protocol (#PID<0.4790.0>) failed to connect: ** (Postgrex.Error) FATAL 28P01 (invalid_password) password authentication failed for user "runner" 565: 16:35:05.678 [error] Postgrex.Protocol (#PID<0.4788.0>) failed to connect: ** (Postgrex.Error) FATAL 28P01 (invalid_password) password authentication failed for user "runner" 566: 16:35:06.783 [error] Postgrex.Protocol (#PID<0.4790.0>) failed to connect: ** (Postgrex.Error) FATAL 28P01 (invalid_password) password authentication failed for user "runner" 567: 16:35:07.606 [error] Postgrex.Protocol (#PID<0.4788.0>) failed to connect: ** (Postgrex.Error) FATAL 28P01 (invalid_password) password authentication failed for user "runner" 568: ** (DBConnection.ConnectionError) connection not available and request was dropped from queue after 5998ms. This means requests are coming in and your connection pool cannot serve them fast enough. You can address this by: 569: 1. Ensuring your database is available and that you can connect to it 570: 2. Tracking down slow queries and making sure they are running fast enough 571: 3. Increasing the pool_size (although this increases resource consumption) 572: 4. Allowing requests to wait longer by increasing :queue_target and :queue_interval 573: See DBConnection.start_link/2 for more information 574: (db_connection 2.8.1) lib/db_connection/ownership.ex:108: DBConnection.Ownership.ownership_checkout/2 575: (ecto_sql 3.13.2) lib/ecto/adapters/sql/sandbox.ex:554: Ecto.Adapters.SQL.Sandbox.checkout/2 576: (ecto_sql 3.13.2) lib/ecto/adapters/sql/sandbox.ex:621: Ecto.Adapters.SQL.Sandbox.unboxed_run/2 577: (ecto_sql 3.13.2) lib/ecto/migrator.ex:170: Ecto.Migrator.with_repo/3 578: (ecto_sql 3.13.2) lib/mix/tasks/ecto.migrate.ex:151: anonymous fn/5 in Mix.Tasks.Ecto.Migrate.run/2 579: (elixir 1.19.2) lib/enum.ex:2520: Enum."-reduce/3-lists^foldl/2-0-"/3 580: (ecto_sql 3.13.2) lib/mix/tasks/ecto.migrate.ex:139: Mix.Tasks.Ecto.Migrate.run/2 581: (mix 1.19.2) lib/mix/task.ex:499: anonymous fn/3 in Mix.Task.run_task/5 582: ##[error]Process completed with exit code 1. 583: Post job cleanup. ... 593: [command]/usr/bin/git config --local --unset-all http.https://github.com/.extraheader 594: [command]/usr/bin/git submodule foreach --recursive sh -c "git config --local --name-only --get-regexp 'http\.https\:\/\/github\.com\/\.extraheader' && git config --local --unset-all 'http.https://github.com/.extraheader' || :" 595: Print service container logs: 208acc20c0fa49b4b1afa728cfd21890_ghcriopgmqpg17pgmqv170_1e1c06 596: ##[command]/usr/bin/docker logs --details c0fbce6365c407ce6706076773b4915e5a3d93ae81d4ec475fc159209fe12492 597: The files belonging to this database system will be owned by user "postgres". 598: initdb: warning: enabling "trust" authentication for local connections 599: initdb: hint: You can change this by editing pg_hba.conf or using the option -A, or --auth-local and --auth-host, the next time you run initdb. 600: This user must also own the server process. 601: 2025-11-09 16:33:28.474 UTC [1] LOG: starting PostgreSQL 17.6 (Debian 17.6-2.pgdg12+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 12.2.0-14+deb12u1) 12.2.0, 64-bit 602: 2025-11-09 16:33:28.474 UTC [1] LOG: listening on IPv4 address "0.0.0.0", port 5432 603: 2025-11-09 16:33:28.474 UTC [1] LOG: listening on IPv6 address "::", port 5432 604: 2025-11-09 16:33:28.475 UTC [1] LOG: listening on Unix socket "/var/run/postgresql/.s.PGSQL.5432" 605: 2025-11-09 16:33:28.478 UTC [51] LOG: database system was shut down at 2025-11-09 16:33:28 UTC 606: 2025-11-09 16:33:28.482 UTC [54] LOG: pg_partman master background worker master process initialized with role postgres 607: 2025-11-09 16:33:28.482 UTC [1] LOG: database system is ready to accept connections 608: 2025-11-09 16:35:04.198 UTC [136] FATAL: password authentication failed for user "runner" 609: 2025-11-09 16:35:04.198 UTC [136] DETAIL: Role "runner" does not exist. 610: Connection matched file "/var/lib/postgresql/data/pg_hba.conf" line 128: "host all all all scram-sha-256" 611: 2025-11-09 16:35:04.198 UTC [137] FATAL: password authentication failed for user "runner" 612: 2025-11-09 16:35:04.198 UTC [137] DETAIL: Role "runner" does not exist. 613: Connection matched file "/var/lib/postgresql/data/pg_hba.conf" line 128: "host all all all scram-sha-256" 614: 2025-11-09 16:35:05.678 UTC [138] FATAL: password authentication failed for user "runner" 615: 2025-11-09 16:35:05.678 UTC [138] DETAIL: Role "runner" does not exist. 616: Connection matched file "/var/lib/postgresql/data/pg_hba.conf" line 128: "host all all all scram-sha-256" 617: 2025-11-09 16:35:06.783 UTC [139] FATAL: password authentication failed for user "runner" 618: 2025-11-09 16:35:06.783 UTC [139] DETAIL: Role "runner" does not exist. 619: Connection matched file "/var/lib/postgresql/data/pg_hba.conf" line 128: "host all all all scram-sha-256" 620: 2025-11-09 16:35:07.606 UTC [140] FATAL: password authentication failed for user "runner" 621: 2025-11-09 16:35:07.606 UTC [140] DETAIL: Role "runner" does not exist.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[mikkihugo]

Greta