functional-core-imperative-shell

函数式核心，命令式外壳（FCIS）

概述

核心原则： 将纯业务逻辑（函数式核心）与副作用（命令式外壳）分离。纯函数放在一个文件中，I/O 操作放在另一个文件中。

为何重要： 纯函数易于测试（无需模拟）。I/O 代码被隔离在精简的外壳中。当业务逻辑没有副作用时，结构上就不可能出现错误。

适用场景

以下情况使用 FCIS：

• 编写任何新代码文件
• 重构现有代码
• 在架构决策中审查代码
• 决定逻辑应放在哪里

触发情形：

• “这个函数应该放在哪里？”
• 创建新文件
• 在逻辑中添加数据库调用
• 在计算中添加文件 I/O
• 编写需要复杂模拟的测试

强制要求：文件分类

必须为每个包含运行时行为的文件添加模式注释：

// pattern: Functional Core
// pattern: Imperative Shell
// pattern: Mixed (needs refactoring)

如果文件确实无法拆分（很少见），请说明原因：

// pattern: Mixed (unavoidable)
// Reason: [specific technical justification]
// Example: Performance-critical path where separating I/O causes unacceptable overhead

任何包含运行时行为的文件都必须分类。 如果你创建的文件包含函数、带方法的类或编排逻辑，却没有添加此注释，就违反了此要求。

豁免：不包含运行时行为的文件

请勿为以下文件添加模式注释：

• 仅包含类型的文件——只导出类型、接口或类型别名的文件（不含运行时代码）
• 常量／枚举类文件——静态数据声明，不含函数
• 汇总／索引文件——仅重新导出（export * from './foo'）
• 测试文件——测试核心／外壳代码，但它们本身既不是核心，也不是外壳
• 生成文件——机器生成的代码
• Bash／Shell 脚本（.sh、.bash）——本质上是命令式的
• 配置文件（eslint.config.js、tsconfig.json、.env 等）
• Markdown 文档（.md）
• HTML 文件（.html）
• 任务运行器文件（justfile、Makefile 等）
• 包清单（package.json、pyproject.toml 等）
• 数据文件（JSON、YAML、CSV 等）

注意： 如果豁免文件后来加入了运行时逻辑（例如，“types”文件中新增了辅助函数，或常量文件中新增了工厂函数），就不再符合豁免条件，必须进行分类。

分类适用于包含运行时行为的应用源文件（包含逻辑的函数、带方法的类、I/O 编排）。

文件类型定义

函数式核心文件

仅包含：

• 纯函数（相同输入始终产生相同输出）
• 业务逻辑、验证、计算、转换
• 数据结构操作
• 日志记录（例外：函数式核心允许使用日志记录器）

绝不包含：

• 文件 I/O（读取、写入文件）
• 数据库操作（查询、更新、连接）
• HTTP 请求或响应
• 环境变量访问
• Date.now()、Math.random() 或其他非确定性函数
• 函数作用域之外的状态变更

日志记录例外： 函数可以接收并使用日志记录器。单元测试时，传入空操作日志记录器。这是函数式核心中唯一允许的副作用。

测试形式： 简单断言，不使用模拟对象（如使用日志记录器，则日志记录器除外）。

命令式外壳文件

仅包含：

• I/O 操作：文件系统、数据库、HTTP、环境
• 编排：收集数据 -> 调用函数式核心 -> 持久化结果
• I/O 故障的错误处理
• 最少量的业务逻辑（仅用于协调）

绝不包含：

• 复杂计算
• 业务规则验证
• 格式转换以外的数据转换

测试形式： 使用真实依赖项或测试替身的集成测试。

代码流模式

1. GATHER (Shell):  Collect data from external sources
2. PROCESS (Core):  Transform input to output (pure)
3. PERSIST (Shell): Save results externally

每项操作都必须遵循此顺序。 无一例外。

决策框架

编写函数前，先问：

digraph fcis_decision {
    "Writing a function" [shape=ellipse];
    "Can run without external dependencies?" [shape=diamond];
    "Does it coordinate I/O?" [shape=diamond];
    "Functional Core" [shape=box, style=filled, fillcolor=lightblue];
    "Imperative Shell" [shape=box, style=filled, fillcolor=lightgreen];
    "STOP: Refactor or escalate" [shape=octagon, style=filled, fillcolor=red, fontcolor=white];

    "Writing a function" -> "Can run without external dependencies?";
    "Can run without external dependencies?" -> "Functional Core" [label="yes"];
    "Can run without external dependencies?" -> "Does it coordinate I/O?" [label="no"];
    "Does it coordinate I/O?" -> "Imperative Shell" [label="yes"];
    "Does it coordinate I/O?" -> "STOP: Refactor or escalate" [label="no"];
}

要问的问题：

• 这段逻辑能否在不使用文件系统、数据库、网络或环境的情况下运行？
  • 是 -> 函数式核心
  • 否 -> 它是否协调 I/O 或包含业务逻辑？
    • 协调 I/O -> 命令式外壳
    • 业务逻辑 + I/O -> 停止。重构或向用户升级处理。

常见错误与自我辩解

借口/思维模式	事实	应对方式
“在这个计算中只读取一个文件”	文件 I/O = 副作用。不属于函数式核心。	提取到外壳中。将数据作为参数传入。
“数据库是作为参数传入的，所以这是纯函数”	数据库操作属于 I/O。不是纯操作。	移到外壳中。核心接收数据，而不是数据库连接。
“此验证需要检查文件是否存在”	检查文件系统 = I/O。不属于函数式核心。	外壳检查文件，并将布尔值传给核心进行验证。
“只是一个小的 HTTP 调用，不会有影响”	HTTP = 副作用。会破坏纯度保证。	外壳发起请求，核心处理响应数据。
“时间戳计算需要 Date.now()”	不确定性。不是纯操作。	外壳将时间戳作为参数传入。
“日志记录是副作用，应该移除”	**错误。**明确允许记录日志。	保留日志记录器。这是例外。
“这个函数同时处理逻辑和 I/O，但这样更简单”	混合关注点 = 不用模拟对象就无法测试。	拆分为核心（逻辑）+ 外壳（I/O）。轻松测试核心。
“文件分类只是额外工作”	它能避免整类错误。不可妥协。	添加分类注释。只需 10 秒。
“我稍后再重构”	以后永远不会来。现在就做。	现在就分类并分离。
“性能要求必须混合处理”	用基准测试证明。通常并非如此。	先分离。用证据优化。将其标记为混合（不可避免），并说明理由。

危险信号 - 停止并重构

如果发现自己在做以下任何一项，立即停止：

• 在“纯”函数中进行文件 I/O（open、read、write、exists 检查）
• 将数据库作为参数传给函数式核心（查询、更新、连接）
• 在业务逻辑中发起 HTTP 请求（fetch、axios、requests）
• 在计算中使用环境变量（process.env、os.getenv）
• 在函数式核心中使用 Math.random() 或 Date.now()（不确定性）
• 创建运行时行为的文件，却没有添加模式分类注释
• 认为“就这一次”可以混合关注点

以上情况都意味着： 将 I/O 提取到外壳中。将数据传给核心。正确分类文件。

实现模式

函数式核心模式

# pattern: Functional Core

def calculate_total_with_tax(items, tax_rate, logger=None):
    """Pure calculation: same inputs always produce same output."""
    if logger:
        logger.debug(f"Calculating total for {len(items)} items")

    subtotal = sum(item['price'] * item['quantity'] for item in items)
    tax = subtotal * tax_rate
    total = subtotal + tax

    return {
        'subtotal': subtotal,
        'tax': tax,
        'total': total
    }

没有 I/O。没有数据库。没有文件系统。只有计算。

命令式外壳模式

# pattern: Imperative Shell

def process_order(order_id, db, logger):
    """Orchestrates: gather -> process -> persist."""

    # GATHER: Collect data from external sources
    items = db.get_order_items(order_id)
    tax_rate = db.get_tax_rate_for_order(order_id)

    # PROCESS: Call Functional Core (pure logic)
    result = calculate_total_with_tax(items, tax_rate, logger)

    # PERSIST: Save results externally
    db.update_order_total(order_id, result['total'])

    return result

外壳应保持精简。核心负责繁重工作。可单独测试。

混合（需要重构）- 错误示例

# pattern: Mixed (needs refactoring)

def calculate_and_save_total(order_id, db):
    """BAD: Mixes calculation with I/O. Hard to test."""
    items = db.get_order_items(order_id)  # I/O
    subtotal = sum(item['price'] for item in items)  # Logic
    tax_rate = db.get_tax_rate_for_order(order_id)  # I/O
    tax = subtotal * tax_rate  # Logic
    total = subtotal + tax  # Logic
    db.update_order_total(order_id, total)  # I/O
    return total

测试此模式需要数据库模拟对象。很脆弱。使用上述模式重构。

日志记录器例外详情

明确允许在函数式核心中使用日志记录器。

# pattern: Functional Core

def validate_order(order_data, logger=None):
    """Pure validation with logging."""
    if logger:
        logger.info(f"Validating order {order_data.get('id')}")

    errors = []

    if not order_data.get('items'):
        errors.append("Order must have items")

    if order_data.get('total', 0) < 0:
        errors.append("Total cannot be negative")

    if logger and errors:
        logger.warning(f"Validation failed: {errors}")

    return {'valid': len(errors) == 0, 'errors': errors}

对于单元测试： 传入空操作日志记录器或 None。函数在测试中仍保持纯粹。

重构模式

分离关注点的常见模式：

从非纯函数中提取纯核心

症状： 函数将 I/O 与逻辑混在一起

# BEFORE - hard to test
def process_order(order_id: str) -> None:
    order = db.fetch(order_id)           # I/O
    discount = calculate_discount(order)  # Pure logic
    total = apply_discount(order, discount)  # Pure logic
    db.save(order_id, total)             # I/O

# AFTER - pure core extracted
def calculate_order_total(order: Order, rules: DiscountRules) -> Decimal:
    """Pure function - easy to test."""
    discount = calculate_discount(order, rules)
    return apply_discount(order, discount)

def process_order(order_id: str) -> None:
    """Thin I/O wrapper."""
    order = db.fetch(order_id)
    total = calculate_order_total(order, get_discount_rules())
    db.save(order_id, total)

返回值，而不是修改状态

症状： 方法就地修改状态，导致难以比较修改前后的结果

# BEFORE - mutation
def sort_tasks(tasks: list[Task]) -> None:
    tasks.sort(key=lambda t: t.priority)

# AFTER - returns new value
def sorted_tasks(tasks: list[Task]) -> list[Task]:
    return sorted(tasks, key=lambda t: t.priority)

添加缺失的逆向操作

症状： 存在单向操作，但没有逆向操作来测试往返过程

# BEFORE - only encode exists
def encode_message(msg: dict) -> bytes:
    return msgpack.packb(msg)

# AFTER - add decode for roundtrip testing
def decode_message(data: bytes) -> dict:
    return msgpack.unpackb(data)

替换硬编码依赖项

症状： 函数使用全局变量或硬编码配置，无法测试边界情况

# BEFORE - uses global
def validate_input(data: str) -> bool:
    return len(data) <= CONFIG.max_length

# AFTER - dependency injected
def validate_input(data: str, max_length: int) -> bool:
    return len(data) <= max_length

重构优先级

模式	影响	工作量	优先级
提取纯核心	高	中	首先处理
添加缺失的逆向操作	高	低	快速见效
返回结果而非修改状态	中	低	轻松改进
注入依赖项	中	中	测试受阻时处理

重构检查清单

发现职责混杂时：

• 识别纯计算（逻辑、计算、验证）
• 将纯代码提取到 Functional Core 文件中
• 识别 I/O 操作（文件、数据库、HTTP、环境）
• 将 I/O 保留在 Imperative Shell 文件中
• Shell 收集数据、调用 Core、持久化结果
• 在两个文件中添加模式分类注释
• 使用简单断言测试 Core（除 logger 外不使用 mock）
• 使用集成测试测试 Shell

如果无法分离： 向用户升级处理，并提供具体的技术理由。不要想当然地认为职责混杂是必要的。

总结

FCIS 的三条规则：

1. Functional Core： 仅包含纯函数。除日志外不进行 I/O。易于测试。
2. Imperative Shell： 仅协调 I/O。逻辑尽可能精简。调用 Core。
3. 对每个具有运行时行为的文件进行分类。 仅包含类型的文件、常量、barrel 文件、测试和生成的文件除外。

不确定时： 它能否在没有外部依赖的情况下运行？→ Functional Core。否则 → Imperative Shell。

日志例外： 可在任何地方使用 logger。单元测试时传入 no-op logger。

职责混杂 = 需要重构。 提取、分离、分类。现在就做，不要拖到以后。