@@ -41,6 +41,29 @@ This finds commits between the last cherry-picked commit and the next merge
commit in the source branch. It stops at the merge commit since that typically
represents a logical grouping of commits (e.g., a pull request).
+To apply the next set of commits using a Claude agent::
+
+ ./tools/pickman/pickman apply us/next
+
+This uses the Claude Agent SDK to automate the cherry-pick process. The agent
+will:
+
+- Run git status to check the repository state
+- Cherry-pick each commit in order
+- Handle simple conflicts automatically
+- Report status after completion
+- Update the database with the last successfully applied commit
+
+Requirements
+------------
+
+To use the ``apply`` command, install the Claude Agent SDK::
+
+ pip install claude-agent-sdk
+
+You will also need an Anthropic API key set in the ``ANTHROPIC_API_KEY``
+environment variable.
+
Database
--------
@@ -34,6 +34,11 @@ def parse_args(argv):
help='Add a source branch to track')
add_source.add_argument('source', help='Source branch name')
+ apply_cmd = subparsers.add_parser('apply',
+ help='Apply next commits using Claude')
+ apply_cmd.add_argument('source', help='Source branch name')
+ apply_cmd.add_argument('-b', '--branch', help='Branch name to create')
+
subparsers.add_parser('compare', help='Compare branches')
subparsers.add_parser('list-sources', help='List tracked source branches')
@@ -15,6 +15,7 @@ our_path = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, os.path.join(our_path, '..'))
# pylint: disable=wrong-import-position,import-error
+from pickman import agent
from pickman import database
from pickman import ftest
from u_boot_pylib import command
@@ -226,6 +227,78 @@ def do_next_set(args, dbs):
return 0
+def do_apply(args, dbs):
+ """Apply the next set of commits using Claude agent
+
+ Args:
+ args (Namespace): Parsed arguments with 'source' and 'branch' attributes
+ dbs (Database): Database instance
+
+ Returns:
+ int: 0 on success, 1 on failure
+ """
+ source = args.source
+ commits, merge_found, error = get_next_commits(dbs, source)
+
+ if error:
+ tout.error(error)
+ return 1
+
+ if not commits:
+ tout.info('No new commits to cherry-pick')
+ return 0
+
+ # Save current branch to return to later
+ original_branch = run_git(['rev-parse', '--abbrev-ref', 'HEAD'])
+
+ # Generate branch name if not provided
+ branch_name = args.branch
+ if not branch_name:
+ # Use first commit's short hash as part of branch name
+ branch_name = f'cherry-{commits[0].short_hash}'
+
+ if merge_found:
+ tout.info(f'Applying next set from {source} ({len(commits)} commits):')
+ else:
+ tout.info(f'Applying remaining commits from {source} '
+ f'({len(commits)} commits, no merge found):')
+
+ tout.info(f' Branch: {branch_name}')
+ for commit in commits:
+ tout.info(f' {commit.short_hash} {commit.subject}')
+ tout.info('')
+
+ # Add commits to database with 'pending' status
+ source_id = dbs.source_get_id(source)
+ for commit in commits:
+ dbs.commit_add(commit.hash, source_id, commit.subject, commit.author,
+ status='pending')
+ dbs.commit()
+
+ # Convert CommitInfo to tuple format expected by agent
+ commit_tuples = [(c.hash, c.short_hash, c.subject) for c in commits]
+ success = agent.cherry_pick_commits(commit_tuples, source, branch_name)
+
+ # Update commit status based on result
+ status = 'applied' if success else 'conflict'
+ for commit in commits:
+ dbs.commit_set_status(commit.hash, status)
+ dbs.commit()
+
+ # Return to original branch
+ current_branch = run_git(['rev-parse', '--abbrev-ref', 'HEAD'])
+ if current_branch != original_branch:
+ tout.info(f'Returning to {original_branch}')
+ run_git(['checkout', original_branch])
+
+ if success:
+ tout.info(f"Use 'pickman commit-source {source} {commits[-1].short_hash}' "
+ 'to update the database')
+
+ return 0 if success else 1
+
+
+
def do_test(args, dbs): # pylint: disable=unused-argument
"""Run tests for this module.
@@ -247,6 +320,7 @@ def do_test(args, dbs): # pylint: disable=unused-argument
# Command dispatch table
COMMANDS = {
'add-source': do_add_source,
+ 'apply': do_apply,
'compare': do_compare,
'list-sources': do_list_sources,
'next-set': do_next_set,
@@ -111,6 +111,20 @@ class TestParseArgs(unittest.TestCase):
self.assertEqual(args.cmd, 'add-source')
self.assertEqual(args.source, 'us/next')
+ def test_parse_apply(self):
+ """Test parsing apply command."""
+ args = pickman.parse_args(['apply', 'us/next'])
+ self.assertEqual(args.cmd, 'apply')
+ self.assertEqual(args.source, 'us/next')
+ self.assertIsNone(args.branch)
+
+ def test_parse_apply_with_branch(self):
+ """Test parsing apply command with branch."""
+ args = pickman.parse_args(['apply', 'us/next', '-b', 'my-branch'])
+ self.assertEqual(args.cmd, 'apply')
+ self.assertEqual(args.source, 'us/next')
+ self.assertEqual(args.branch, 'my-branch')
+
def test_parse_compare(self):
"""Test parsing compare command."""
args = pickman.parse_args(['compare'])
@@ -816,5 +830,115 @@ class TestNextSet(unittest.TestCase):
self.assertIn('bbb222b Second commit', output)
+class TestGetNextCommits(unittest.TestCase):
+ """Tests for get_next_commits function."""
+
+ def setUp(self):
+ """Set up test fixtures."""
+ fd, self.db_path = tempfile.mkstemp(suffix='.db')
+ os.close(fd)
+ os.unlink(self.db_path)
+ self.old_db_fname = control.DB_FNAME
+ control.DB_FNAME = self.db_path
+ database.Database.instances.clear()
+
+ def tearDown(self):
+ """Clean up test fixtures."""
+ control.DB_FNAME = self.old_db_fname
+ if os.path.exists(self.db_path):
+ os.unlink(self.db_path)
+ database.Database.instances.clear()
+ command.TEST_RESULT = None
+
+ def test_get_next_commits_source_not_found(self):
+ """Test get_next_commits with unknown source"""
+ with terminal.capture():
+ dbs = database.Database(self.db_path)
+ dbs.start()
+ commits, merge_found, error = control.get_next_commits(dbs,
+ 'unknown')
+ self.assertIsNone(commits)
+ self.assertFalse(merge_found)
+ self.assertIn('not found', error)
+ dbs.close()
+
+ def test_get_next_commits_with_merge(self):
+ """Test get_next_commits finding commits up to merge"""
+ with terminal.capture():
+ dbs = database.Database(self.db_path)
+ dbs.start()
+ dbs.source_set('us/next', 'abc123')
+ dbs.commit()
+
+ log_output = (
+ 'aaa111|aaa111a|Author 1|First commit|abc123\n'
+ 'bbb222|bbb222b|Author 2|Merge branch|aaa111 ccc333\n'
+ )
+ command.TEST_RESULT = command.CommandResult(stdout=log_output)
+
+ commits, merge_found, error = control.get_next_commits(dbs,
+ 'us/next')
+ self.assertIsNone(error)
+ self.assertTrue(merge_found)
+ self.assertEqual(len(commits), 2)
+ self.assertEqual(commits[0].short_hash, 'aaa111a')
+ self.assertEqual(commits[1].short_hash, 'bbb222b')
+ dbs.close()
+
+
+class TestApply(unittest.TestCase):
+ """Tests for apply command."""
+
+ def setUp(self):
+ """Set up test fixtures."""
+ fd, self.db_path = tempfile.mkstemp(suffix='.db')
+ os.close(fd)
+ os.unlink(self.db_path)
+ self.old_db_fname = control.DB_FNAME
+ control.DB_FNAME = self.db_path
+ database.Database.instances.clear()
+
+ def tearDown(self):
+ """Clean up test fixtures."""
+ control.DB_FNAME = self.old_db_fname
+ if os.path.exists(self.db_path):
+ os.unlink(self.db_path)
+ database.Database.instances.clear()
+ command.TEST_RESULT = None
+
+ def test_apply_source_not_found(self):
+ """Test apply with unknown source"""
+ with terminal.capture():
+ dbs = database.Database(self.db_path)
+ dbs.start()
+ dbs.close()
+
+ database.Database.instances.clear()
+
+ args = argparse.Namespace(cmd='apply', source='unknown')
+ with terminal.capture() as (_, stderr):
+ ret = control.do_pickman(args)
+ self.assertEqual(ret, 1)
+ self.assertIn("Source 'unknown' not found", stderr.getvalue())
+
+ def test_apply_no_commits(self):
+ """Test apply with no new commits"""
+ with terminal.capture():
+ dbs = database.Database(self.db_path)
+ dbs.start()
+ dbs.source_set('us/next', 'abc123')
+ dbs.commit()
+ dbs.close()
+
+ database.Database.instances.clear()
+ command.TEST_RESULT = command.CommandResult(stdout='')
+
+ args = argparse.Namespace(cmd='apply', source='us/next')
+ with terminal.capture() as (stdout, _):
+ ret = control.do_pickman(args)
+ self.assertEqual(ret, 0)
+ self.assertIn('No new commits to cherry-pick', stdout.getvalue())
+
+
if __name__ == '__main__':
unittest.main()