Provenance-Aware Task Decomposition

The problem

Breaking down a large goal into smaller steps can lead to errors or unauthorized changes as the plan progresses. It is difficult to ensure that every sub-task remains accurate and untampered with throughout the process.

What it does

It breaks a large goal into a sequence of smaller tasks and assigns a unique digital fingerprint to every single step. This ensures that each part of the plan remains consistent and verified as it is being built.

Why it matters

It ensures that a complex plan remains accurate and secure from start to finish.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 task_planner.py
Decomposition Structure:
* Project (SHA-256: 814a4723...) 
  * Plan (SHA-256: c3ecc74c...) 
    * DefineScope (SHA-256: 383ba50e...) 
    * EstimateResources (SHA-256: cdfe6b7a...) 
  * Execute (SHA-256: 413e8a43...) 
    * WriteCode (SHA-256: 61205336...) 
    * TestCode (SHA-256: d1b79ec1...) 
  * Verify (SHA-256: 54f1a337...) 
    * Review (SHA-256: eb2fb1ce...) 
    * Deploy (SHA-256: 81a19a53...) 

Integrity Verification:
{'Project': '814a47233250f3f920013e1afee4b3a4cf5f72ca7816beb593002fee4f136509', 'Plan': 'c3ecc74c0ac053a1de7c04efbca477aba872b495ceaf66f0d8487003fada9dfc', 'DefineScope': '383ba50e8d0dc15cf10110df065247d5d8e720c74ec72acfc96c57bb10ec2a04', 'EstimateResources': 'cdfe6b7a59ed5297b961219f1fbfcea20da9a298e02b278b1428891923fefd94', 'Execute': '413e8a4375edc351b16cbc1dc45744dc385c2d91724fd5d1dbf5b672a435dc43', 'WriteCode': '6120533685b5511b2e507a9f3eb5d59b82b90cc43e28e7313bf86f0c90000a8b', 'TestCode': 'd1b79ec10a9318505e8fe317e37985e4a7d0ba3b1295ca3e395812101ca71c9e', 'Verify': '54f1a33700af136a642eba2ba48f6f7712f2d70d188a7aa27704a7ffdb3580ff', 'Review': 'eb2fb1ceb25a34af01959c2e5c6d2dcd204520e9ab83026b947d2dbac9a53041', 'Deploy': '81a19a5308c948135c4efb5fd16ff1ffb4

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 96 lines, one file, standard library only.

import hashlib
from dataclasses import dataclass
from typing import List, Dict, Optional
import sys

def _generate_hash(name: str, description: str, parent_hash: Optional[str] = None) -> str:
    input_data = f"{name}{description}{parent_hash or ''}".encode()
    return hashlib.sha256(input_data).hexdigest()

class Subtask:
    def __init__(self, name: str, description: str, parent_hash: Optional[str] = None):
        self.name = name
        self.description = description
        self.parent_hash = parent_hash
        self.children = []
        self.hash = _generate_hash(name, description, parent_hash)
        self.tamper_detected = False

    def add_child(self, child: 'Subtask') -> None:
        self.children.append(child)
        # Intentionally not updating self.hash to maintain immutability

    def verify(self, parent_hash: Optional[str] = None) -> bool:
        """
        Returns True if:
        1. Own hash matches computed hash
        2. Parent hash matches provided parent_hash (if any)
        """
        # Check own hash integrity
        if self.hash != _generate_hash(self.name, self.description, self.parent_hash):
            self.tamper_detected = True
            return False
        
        # Check parent hash integrity if parent provided
        if parent_hash is not None and self.parent_hash != parent_hash:
            self.tamper_detected = True
            return False
        
        self.tamper_detected = False
        return True

class HTNPlanner:
    def __init__(self):
        self.root_task = None

    def decompose(self, root_name: str, root_description: str, decomposition_rules: Dict[str, List[str]]) -> 'Subtask':
        self.root_task = Subtask(root_name, root_description)
        self._decompose_task(self.root_task, decomposition_rules)
        return self.root_task

    def _decompose_task(self, task: Subtask, decomposition_rules: Dict[str, List[str]]) -> None:
        if task.name in decomposition_rules:
            for subtask_name in decomposition_rules[task.name]:
                subtask = Subtask(
                    subtask_name,
                    f"Subtask {subtask_name} under {task.name}",
                    parent_hash=task.hash
                )
                task.add_child(subtask)
                self._decompose_task(subtask, decomposition_rules)

    def verify_integrity(self, task: Subtask, parent_hash: Optional[str] = None) -> Dict:
        results = {task.name: {
            'hash': task.hash,
            'tamper_detected': not task.verify(parent_hash)
        }}
        for child in task.children:
            results.update(self.verify_integrity(child, task.hash))
        return results

if __name__ == '__main__':
    decomposition_rules = {
        'Project': ['Plan', 'Execute', 'Verify'],
        'Plan': ['DefineScope', 'EstimateResources'],
        'Execute': ['WriteCode', 'TestCode'],
        'Verify': ['Review', 'Deploy']
    }

    planner = HTNPlanner()
    root_task = planner.decompose('Project', 'Sample project with provenance tracking', decomposition_rules)

    print('Decomposition Structure:')
    def print_tree(task, indent='', parent_hash=None):
        verify_result = task.verify(parent_hash)
        status = 'TAINTED' if not verify_result else 'INTACT'
        print(f"{indent}* {task.name} (SHA-256: {task.hash[:8]}...) [{status}]")
        for child in task.children:
            print_tree(child, indent + ' ', task.hash)
    
    print_tree(root_task)

    # Simulate tampering by altering a task's name
    root_task.children[0].name = "HackedPlan"  # Original name was 'Plan'

    print('\nIntegrity Verification after tampering:')
    print(planner.verify_integrity(root_task))