Internals: How Linear 2.0 Manages 2026 Project Workflows for Engineering Teams

use std::collections::{HashMap, HashSet};
use thiserror::Error;

/// Custom error type for Workflow DAG operations
#[derive(Error, Debug, PartialEq)]
pub enum DagError {
    #[error(\"Node {0} already exists in DAG\")]
    DuplicateNode(String),
    #[error(\"Edge from {0} to {1} would create a cycle\")]
    CycleDetected(String, String),
    #[error(\"Node {0} not found in DAG\")]
    MissingNode(String),
    #[error(\"Maximum DAG depth {0} exceeded\")]
    DepthExceeded(usize),
}

/// Represents a single node in the project workflow DAG
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowNode {
    pub id: String,
    pub node_type: NodeType,
    pub metadata: HashMap,
    pub max_depth: usize,
}

#[derive(Debug, Clone, PartialEq)]
pub enum NodeType {
    Task,
    Milestone,
    Epic,
    ApprovalGate,
    AutomationHook,
}

/// Directed Acyclic Graph for managing 2026 project workflows
#[derive(Debug, Default)]
pub struct WorkflowDag {
    nodes: HashMap,
    edges: HashMap>, // adjacency list: source -> targets
    reverse_edges: HashMap>, // target -> sources
}

impl WorkflowDag {
    /// Create a new empty WorkflowDag
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a node to the DAG, returns error if node already exists
    pub fn add_node(&mut self, node: WorkflowNode) -> Result<(), DagError> {
        if self.nodes.contains_key(&node.id) {
            return Err(DagError::DuplicateNode(node.id));
        }
        self.nodes.insert(node.id.clone(), node);
        self.edges.entry(node.id.clone()).or_default();
        self.reverse_edges.entry(node.id.clone()).or_default();
        Ok(())
    }

    /// Add a directed edge from source to target, validates no cycles
    pub fn add_edge(&mut self, source_id: &str, target_id: &str) -> Result<(), DagError> {
        // Check both nodes exist
        if !self.nodes.contains_key(source_id) {
            return Err(DagError::MissingNode(source_id.to_string()));
        }
        if !self.nodes.contains_key(target_id) {
            return Err(DagError::MissingNode(target_id.to_string()));
        }
        // Check if edge already exists
        if self.edges.get(source_id).map_or(false, |targets| targets.contains(target_id)) {
            return Ok(()); // idempotent
        }
        // Temporarily add edge to check for cycles
        self.edges.get_mut(source_id).unwrap().insert(target_id.to_string());
        self.reverse_edges.get_mut(target_id).unwrap().insert(source_id.to_string());
        // Validate no cycles
        if self.detect_cycle() {
            // Roll back
            self.edges.get_mut(source_id).unwrap().remove(target_id);
            self.reverse_edges.get_mut(target_id).unwrap().remove(source_id);
            return Err(DagError::CycleDetected(source_id.to_string(), target_id.to_string()));
        }
        // Check depth constraints
        let target_depth = self.calculate_depth(target_id)?;
        let source_node = self.nodes.get(source_id).unwrap();
        if target_depth > source_node.max_depth {
            // Roll back
            self.edges.get_mut(source_id).unwrap().remove(target_id);
            self.reverse_edges.get_mut(target_id).unwrap().remove(source_id);
            return Err(DagError::DepthExceeded(source_node.max_depth));
        }
        Ok(())
    }

    /// Detect if the DAG has a cycle using DFS
    fn detect_cycle(&self) -> bool {
        let mut visited = HashSet::new();
        let mut recursion_stack = HashSet::new();
        for node_id in self.nodes.keys() {
            if self.dfs_cycle_check(node_id, &mut visited, &mut recursion_stack) {
                return true;
            }
        }
        false
    }

    fn dfs_cycle_check(&self, node_id: &str, visited: &mut HashSet, recursion_stack: &mut HashSet) -> bool {
        if recursion_stack.contains(node_id) {
            return true;
        }
        if visited.contains(node_id) {
            return false;
        }
        visited.insert(node_id.to_string());
        recursion_stack.insert(node_id.to_string());
        if let Some(targets) = self.edges.get(node_id) {
            for target in targets {
                if self.dfs_cycle_check(target, visited, recursion_stack) {
                    return true;
                }
            }
        }
        recursion_stack.remove(node_id);
        false
    }

    /// Calculate the depth of a node (longest path from root)
    fn calculate_depth(&self, node_id: &str) -> Result {
        let mut memo = HashMap::new();
        self.dfs_depth(node_id, &mut memo)
    }

    fn dfs_depth(&self, node_id: &str, memo: &mut HashMap) -> Result {
        if let Some(&depth) = memo.get(node_id) {
            return Ok(depth);
        }
        let sources = self.reverse_edges.get(node_id).unwrap();
        if sources.is_empty() {
            memo.insert(node_id.to_string(), 0);
            return Ok(0);
        }
        let mut max_depth = 0;
        for source in sources {
            let source_depth = self.dfs_depth(source, memo)?;
            max_depth = max_depth.max(source_depth + 1);
        }
        memo.insert(node_id.to_string(), max_depth);
        Ok(max_depth)
    }
}

use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Serialize, Deserialize};
use thiserror::Error;

/// Version vector for tracking causal ordering of mutations
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct VersionVector {
    pub node_id: String,
    pub counters: HashMap,
}

impl VersionVector {
    pub fn new(node_id: &str) -> Self {
        let mut counters = HashMap::new();
        counters.insert(node_id.to_string(), 0);
        Self {
            node_id: node_id.to_string(),
            counters,
        }
    }

    pub fn increment(&mut self) {
        let counter = self.counters.entry(self.node_id.clone()).or_insert(0);
        *counter += 1;
    }

    /// Compare two version vectors: returns true if self is strictly after other
    pub fn happens_after(&self, other: &VersionVector) -> bool {
        let mut all_le = true;
        for (node, &other_count) in &other.counters {
            let self_count = self.counters.get(node).unwrap_or(&0);
            if self_count < other_count {
                return false;
            }
            if self_count > other_count {
                all_le = false;
            }
        }
        !all_le && self.counters.keys().all(|k| other.counters.contains_key(k) || self.counters.get(k).unwrap_or(&0) >= &0)
    }
}

/// Represents a single task in a 2026 project workflow
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Task {
    pub id: String,
    pub title: String,
    pub status: TaskStatus,
    pub assigned_to: Option,
    pub last_updated: u64, // unix ms
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum TaskStatus {
    Backlog,
    InProgress,
    InReview,
    Done,
    Blocked,
}

/// CRDT-based project state store for Linear 2.0
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ProjectState {
    pub project_id: String,
    pub tasks: HashMap,
    pub version: VersionVector,
    pub workflow_dag: WorkflowDag, // from previous snippet, assume it's imported
}

#[derive(Error, Debug, PartialEq)]
pub enum StateError {
    #[error(\"Task {0} not found in project {1}\")]
    TaskNotFound(String, String),
    #[error(\"Invalid status transition for task {0}: {1:?} -> {2:?}\")]
    InvalidTransition(String, TaskStatus, TaskStatus),
    #[error(\"Causal dependency missing: version vector mismatch\")]
    CausalDependencyMissing,
}

impl ProjectState {
    pub fn new(project_id: &str, node_id: &str) -> Self {
        Self {
            project_id: project_id.to_string(),
            tasks: HashMap::new(),
            version: VersionVector::new(node_id),
            workflow_dag: WorkflowDag::new(),
        }
    }

    /// Apply a mutation to the project state, validates causal ordering
    pub fn apply_mutation(&mut self, mutation: StateMutation, local_node_id: &str) -> Result<(), StateError> {
        // Validate causal ordering: mutation's version must be <= current version
        if !self.version.happens_after(&mutation.base_version) && self.version != mutation.base_version {
            return Err(StateError::CausalDependencyMissing);
        }
        // Apply the mutation
        match mutation.mutation_type {
            MutationType::UpdateTask { task_id, new_status, assignee } => {
                let task = self.tasks.get_mut(&task_id).ok_or_else(|| StateError::TaskNotFound(task_id.clone(), self.project_id.clone()))?;
                // Validate status transition against workflow DAG
                if !self.workflow_dag.validate_transition(&task.status, &new_status) {
                    return Err(StateError::InvalidTransition(task_id, task.status.clone(), new_status));
                }
                task.status = new_status;
                task.assigned_to = assignee;
                task.last_updated = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
            }
            MutationType::AddTask { task } => {
                self.tasks.insert(task.id.clone(), task);
            }
        }
        // Merge version vectors
        self.merge_version(mutation.mutation_version);
        Ok(())
    }

    /// Merge another project state into this one (CRDT merge)
    pub fn merge(&mut self, other: &ProjectState) {
        // Merge version vectors
        self.merge_version(other.version.clone());
        // Merge tasks: last write wins by last_updated timestamp
        for (task_id, other_task) in &other.tasks {
            match self.tasks.get_mut(task_id) {
                Some(self_task) => {
                    if other_task.last_updated > self_task.last_updated {
                        *self_task = other_task.clone();
                    }
                }
                None => {
                    self.tasks.insert(task_id.clone(), other_task.clone());
                }
            }
        }
        // Merge workflow DAGs (simplified: union of nodes/edges)
        // Note: Real Linear 2.0 uses a more sophisticated DAG merge CRDT
        for (node_id, node) in &other.workflow_dag.nodes {
            if !self.workflow_dag.nodes.contains_key(node_id) {
                let _ = self.workflow_dag.add_node(node.clone());
            }
        }
        for (source, targets) in &other.workflow_dag.edges {
            for target in targets {
                let _ = self.workflow_dag.add_edge(source, target);
            }
        }
    }

    fn merge_version(&mut self, other_version: VersionVector) {
        for (node, &other_count) in &other_version.counters {
            let self_count = self.version.counters.entry(node.clone()).or_insert(0);
            *self_count = (*self_count).max(other_count);
        }
        self.version.increment();
    }
}

/// Represents a single state mutation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StateMutation {
    pub base_version: VersionVector,
    pub mutation_version: VersionVector,
    pub mutation_type: MutationType,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MutationType {
    UpdateTask {
        task_id: String,
        new_status: TaskStatus,
        assignee: Option,
    },
    AddTask {
        task: Task,
    },
}

use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::time::{Duration, Instant};
use thiserror::Error;
use uuid::Uuid;

/// Edge-deployed workflow validator for Linear 2.0 (2026 architecture)
#[derive(Debug, Clone)]
pub struct EdgeWorkflowValidator {
    pub team_id: String,
    pub rate_limit_per_sec: u32,
    pub max_concurrent_mutations: usize,
    pub allowed_workflows: HashMap, // workflow_id -> rules
    mutation_counts: HashMap, // node_id -> (count, reset_time)
    active_mutations: HashMap, // mutation_id -> start_time
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowRules {
    pub allowed_statuses: Vec,
    pub required_approvals: usize,
    pub max_blocked_tasks: usize,
    pub automation_hooks: Vec,
}

#[derive(Error, Debug, PartialEq)]
pub enum ValidationError {
    #[error(\"Rate limit exceeded for node {0}: {1} mutations/sec (max {2})\")]
    RateLimitExceeded(String, u32, u32),
    #[error(\"Max concurrent mutations exceeded: {0} active (max {1})\")]
    ConcurrentMutationLimit(usize, usize),
    #[error(\"Task {0} status {1:?} not allowed in workflow {2}\")]
    InvalidStatus(String, TaskStatus, String),
    #[error(\"Task {0} requires {1} approvals, has {2}\")]
    InsufficientApprovals(String, usize, usize),
    #[error(\"Workflow {0} has {1} blocked tasks (max {2})\")]
    BlockedTaskLimit(String, usize, usize),
    #[error(\"Mutation timeout: {0}ms exceeded\")]
    MutationTimeout(u64),
}

impl EdgeWorkflowValidator {
    pub fn new(team_id: &str, rate_limit_per_sec: u32, max_concurrent_mutations: usize) -> Self {
        Self {
            team_id: team_id.to_string(),
            rate_limit_per_sec,
            max_concurrent_mutations,
            allowed_workflows: HashMap::new(),
            mutation_counts: HashMap::new(),
            active_mutations: HashMap::new(),
        }
    }

    /// Register a workflow with its rules
    pub fn register_workflow(&mut self, workflow_id: &str, rules: WorkflowRules) {
        self.allowed_workflows.insert(workflow_id.to_string(), rules);
    }

    /// Validate a mutation before applying to the project state
    pub async fn validate_mutation(
        &mut self,
        mutation: &StateMutation,
        project_state: &ProjectState,
        node_id: &str,
    ) -> Result<(), ValidationError> {
        // 1. Check rate limits
        self.check_rate_limit(node_id).await?;

        // 2. Check concurrent mutation limit
        self.active_mutations.retain(|_, &mut start| start.elapsed() < Duration::from_secs(5));
        if self.active_mutations.len() >= self.max_concurrent_mutations {
            return Err(ValidationError::ConcurrentMutationLimit(
                self.active_mutations.len(),
                self.max_concurrent_mutations,
            ));
        }
        let mutation_id = Uuid::new_v4().to_string();
        self.active_mutations.insert(mutation_id.clone(), Instant::now());

        // 3. Check workflow rules
        let workflow_id = &project_state.project_id; // simplified: project_id == workflow_id
        let rules = self.allowed_workflows.get(workflow_id).ok_or_else(|| {
            ValidationError::InvalidStatus(
                \"unknown\".to_string(),
                TaskStatus::Backlog,
                workflow_id.clone(),
            )
        })?;

        // 4. Validate status transition if it's an update task mutation
        if let MutationType::UpdateTask { task_id, new_status, .. } = &mutation.mutation_type {
            if !rules.allowed_statuses.contains(new_status) {
                return Err(ValidationError::InvalidStatus(
                    task_id.clone(),
                    new_status.clone(),
                    workflow_id.clone(),
                ));
            }

            // Check blocked task limit
            let blocked_count = project_state.tasks.values().filter(|t| t.status == TaskStatus::Blocked).count();
            if blocked_count >= rules.max_blocked_tasks {
                return Err(ValidationError::BlockedTaskLimit(
                    workflow_id.clone(),
                    blocked_count,
                    rules.max_blocked_tasks,
                ));
            }

            // Check approvals if moving to Done
            if *new_status == TaskStatus::Done {
                let task = project_state.tasks.get(task_id).ok_or_else(|| {
                    ValidationError::InvalidStatus(task_id.clone(), new_status.clone(), workflow_id.clone())
                })?;
                // Simplified: assume approvals are stored in metadata
                let approvals = task.metadata.get(\"approvals\").unwrap_or(&\"0\".to_string()).parse().unwrap_or(0);
                if approvals < rules.required_approvals {
                    return Err(ValidationError::InsufficientApprovals(
                        task_id.clone(),
                        rules.required_approvals,
                        approvals,
                    ));
                }
            }
        }

        // 5. Check mutation timeout (simplified: 5s max)
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
        let mutation_time = mutation.base_version.counters.values().sum::(); // simplified timestamp
        if now - mutation_time > 5000 {
            return Err(ValidationError::MutationTimeout(now - mutation_time));
        }

        // Clean up active mutation (simplified: in real code, this is done async)
        self.active_mutations.remove(&mutation_id);
        Ok(())
    }

    async fn check_rate_limit(&mut self, node_id: &str) -> Result<(), ValidationError> {
        let now = Instant::now();
        let (count, reset_time) = self.mutation_counts.entry(node_id.to_string()).or_insert((0, now));
        if now.duration_since(*reset_time) > Duration::from_secs(1) {
            *count = 0;
            *reset_time = now;
        }
        *count += 1;
        if *count > self.rate_limit_per_sec {
            return Err(ValidationError::RateLimitExceeded(
                node_id.to_string(),
                *count,
                self.rate_limit_per_sec,
            ));
        }
        Ok(())
    }
}

linear workflow validate --dag-path ./project-dag.json --max-depth 10

impl VersionVector {
    pub fn prune_stale(&mut self, stale_threshold_secs: u64) {
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
        self.counters.retain(|_, &mut count| {
            // Simplified: assume count timestamp is stored elsewhere
            now - (count * 1000) < stale_threshold_secs
        });
    }
}

curl -X POST https://api.linear.app/v2/teams/TEAM_ID/edge-validators \
  -H \"Authorization: Bearer $LINEAR_API_KEY\" \
  -d '{\"region\": \"us-east-1\", \"rate_limit_per_sec\": 15000, \"max_concurrent_mutations\": 200}'